Add the `ClientEncryption.createEncryptedCollection` helper method #1079

stIncMale · 2023-02-04T01:14:59Z

Links to the relevant spec changes with their status are gathered and being kept updated in this comment. Note especially that some tests that you may want to see for this change are yet to be described in the spec.

JAVA-4679

driver-core/src/main/com/mongodb/client/model/CreateCollectionOptions.java

stIncMale · 2023-02-04T01:48:03Z

...streams/src/main/com/mongodb/reactivestreams/client/internal/vault/ClientEncryptionImpl.java

@@ -183,6 +198,85 @@ public Publisher<RewrapManyDataKeyResult> rewrapManyDataKey(final Bson filter, f
                        }));
    }

+    @Override
+    public Publisher<BsonDocument> createEncryptedCollection(final MongoDatabase database, final String collectionName,


@rozza, your knowledge of the Project Reactor and the reactive streams programming is being needed when reviewing this method.

The intended behavior of the implementation of this method can be peeked either in the synchronous implementation, or in the spec: Create Encrypted Collection Helper.

The thing that made this implementation really difficult for me (besides the lack of knowledge) is:

if either createDataKey or createCollection fails, the method must attach to the onError signal the (possibly partially) updated encryptedFields that it would have produced in case of a success.

I achieved the needed behavior by capturing dataKeyMayBeCreated and actualEncryptedFields in the returned Publisher. As far as I understand, the Publisher is constructed such that there are no data races when updating the actualEncryptedFields document (updates are done to the unrelated field documents, which are different elements of a BSON array), and all reads of the updated actualEncryptedFields are ordered after (in the HB order) the updates due to the combination of then and defer methods.

Even if this implementation is correct (at the very least I doubt it is idiomatic), it can't work if there are more than one Subscriber (regardless of whether they exist concurrently with each other or not), which is why I had make the result a one-shot publisher.

stIncMale · 2023-02-04T01:54:02Z

...streams/src/main/com/mongodb/reactivestreams/client/internal/vault/ClientEncryptionImpl.java

+                                .map(dataKey -> new SimpleImmutableEntry<>(field, dataKey))
+                        )
+                        .iterator();
+                Flux<SimpleImmutableEntry<BsonDocument, BsonBinary>> publisherOfFieldsAndDataKeyIds = Flux.concat(publishersOfFieldsAndDataKeyIds)


As a result of Flux.concat all data keys are created sequentially. This way we stop as soon as there is an error, instead possibly continuing to invoke createDataKey despite there already being an error.

stIncMale · 2023-02-04T01:59:55Z

...streams/src/main/com/mongodb/reactivestreams/client/internal/vault/ClientEncryptionImpl.java

+                        .doOnError(RuntimeException.class, e -> {
+                            if (dataKeyMayBeCreated.get()) {
+                                throw new MongoUpdatedEncryptedFieldsException(actualEncryptedFields, format("Failed to create %s.", namespace), e);
+                            } else {
+                                throw e;
+                            }
+                        });


Exceptions from doOnError/defer are handled by the Project Reactor such that the reactive streams protocol is not violated. I am not sure what is a better / more idiomatic way of transforming the error.

P.S. While writing this comment, I tried to find a better way again and found Flux.handle, but it can't help because it is for handling emitted items, not errors.

Note that doOnError suppresses the original e when we throw MongoUpdatedEncryptedFieldsException, instead of treating the thrown exception as a replacement for the original one. As a result, we have e as both the cause of MongoUpdatedEncryptedFieldsException and an element in the suppressed set, which causes ugly stack-traces and implies that we are abusing doOnError. However, it may be that the Reactor simply does not support what we need.

@rozza Thank you for pointing out the onErrorMap method, the emitted exceptions are nice now. There is no excuse for how I failed to find it myself.

stIncMale · 2023-02-04T02:08:47Z

driver-sync/src/main/com/mongodb/client/internal/ClientEncryptionImpl.java

+            if (!fields.isArray()) {
+                throw new MongoConfigurationException(format("`encryptedFields` is incorrectly configured for the collection %s."
+                        + " `encryptedFields.fields` must be an array, but is %s type.", namespace, fields.getBsonType()));


This exception is not according to the current spec, and comes from the following:

https://mongodb.slack.com/archives/C0406ECL478/p1675269046945409

https://jira.mongodb.org/browse/DRIVERS-2540 (does not have a completed spec change), which proposes to throw in the createCollection methods. And I think that if createCollection methods need to throw, so should the createEncryptedCollection method.

If there is a concern that we should not throw this exception until (if ever) it is approved as a spec change, I will undo.

I undone this part because I realized that the proposal in DRIVERS-2540 is not fully-baked.

stIncMale · 2023-02-04T04:42:25Z

...streams/src/main/com/mongodb/reactivestreams/client/internal/vault/ClientEncryptionImpl.java

+        return OneShotPublisher.from(Mono.defer(() -> {
+            MongoNamespace namespace = new MongoNamespace(database.getName(), collectionName);
+            Bson rawEncryptedFields = createCollectionOptions.getEncryptedFields();
+            if (rawEncryptedFields == null) {
+                throw new MongoConfigurationException(format("`encryptedFields` is not configured for the collection %s.", namespace));


Instead of throwing here, it is likely more idiomatic to return Mono.error, given that we are in the Supplier<Mono> that is passed to Mono.defer. However, if we are to follow this approach consistently, the whole Supplier<Mono> passed to Mono.defer should be wrapped in a try statement that catches all non-fatal exceptions and returns Mono.error instead of throwing them.

But I see no reason to do that:

it is one more thing we should always remember to do;

it arguably makes the code less readable;

why do that when the Project Reactor already wraps the Supplier<Mono> passed to Mono.defer in the try statement and signals onSubscribe followed by onError for us?

I would do this work outside of the construction of the publisher. That way it follows the general validation eg notNull checks and reduces the code in the callable.

I initially had it like that, but then noticed that the other MongoConfigurationException we throw if fields is not an array, should be signalled via the publisher: we need to encode the encryptedFields config to BsonDocument, and this does not seem anymore like just an argument check.

Reporting MongoConfigurationException from the same method in two different ways (throw or signal onError) seems weird, so I decided to also signal the first MongoConfigurationException that you commented on. Given this consideration, does the code seem better now?

Based on your experimental code here, I infer that you still think that failures to render Bson should be treated as fatal and not signalled via the returned publisher. So I did this work outside of the construction of the publisher as you originally requested.

JAVA-4679

rozza

Added a few comments.

I think there may be a more "functional" way with the reactive streams approach. But I liked the procedural nature of it and how it follows the sync version 👍

driver-sync/src/main/com/mongodb/client/vault/ClientEncryption.java

rozza · 2023-02-06T16:35:28Z

driver-sync/src/main/com/mongodb/client/internal/ClientEncryptionImpl.java

+            if (!fields.isArray()) {
+                throw new MongoConfigurationException(format("`encryptedFields` is incorrectly configured for the collection %s."
+                        + " `encryptedFields.fields` must be an array, but is %s type.", namespace, fields.getBsonType()));


driver-sync/src/main/com/mongodb/client/internal/ClientEncryptionImpl.java

driver-core/src/main/com/mongodb/client/model/CreateCollectionOptions.java

driver-sync/src/main/com/mongodb/client/internal/ClientEncryptionImpl.java

rozza · 2023-02-06T17:16:11Z

...streams/src/main/com/mongodb/reactivestreams/client/internal/vault/ClientEncryptionImpl.java

+        return OneShotPublisher.from(Mono.defer(() -> {
+            MongoNamespace namespace = new MongoNamespace(database.getName(), collectionName);
+            Bson rawEncryptedFields = createCollectionOptions.getEncryptedFields();
+            if (rawEncryptedFields == null) {
+                throw new MongoConfigurationException(format("`encryptedFields` is not configured for the collection %s.", namespace));


I would do this work outside of the construction of the publisher. That way it follows the general validation eg notNull checks and reduces the code in the callable.

config/spotbugs/exclude.xml

driver-core/src/main/com/mongodb/client/model/CreateCollectionOptions.java

driver-core/src/main/com/mongodb/client/model/CreateEncryptedCollectionParams.java

katcharov · 2023-02-06T23:50:22Z

driver-sync/src/main/com/mongodb/client/internal/ClientEncryptionImpl.java

+                dataKeyOptions.masterKey(masterKey);
+            }
+            String keyIdBsonKey = "keyId";
+            // any mutable non-thread-safe Boolean should do


I do not understand this comment. What is the non-thread-safe boolean?

Updated. Do you find the updated text clearer?

driver-sync/src/main/com/mongodb/client/internal/ClientEncryptionImpl.java

JAVA-4679

stIncMale · 2023-02-07T01:36:04Z

@rozza

I think there may be a more "functional" way with the reactive streams approach.

What is that approach? I am most of all interested in how to attach the data key IDs to MongoUpdatedEncryptedFieldsException with a functional approach?

JAVA-4679

katcharov

lgtm

rozza · 2023-02-07T17:15:09Z

@rozza

I think there may be a more "functional" way with the reactive streams approach.

What is that approach? I am most of all interested in how to attach the data key IDs to MongoUpdatedEncryptedFieldsException with a functional approach?

Me and my big mouth - turned out to be a fun exercise! Here is my attempt of a more functional approach:

 @Override
    public Publisher<BsonDocument> createEncryptedCollection(final MongoDatabase database, final String collectionName,
            final CreateCollectionOptions createCollectionOptions, final CreateEncryptedCollectionParams createEncryptedCollectionParams) {
        notNull("collectionName", collectionName);
        notNull("createCollectionOptions", createCollectionOptions);
        notNull("createEncryptedCollectionParams", createEncryptedCollectionParams);

        MongoNamespace namespace = new MongoNamespace(database.getName(), collectionName);
        Bson rawEncryptedFields = createCollectionOptions.getEncryptedFields();
        if (rawEncryptedFields == null) {
            throw new MongoConfigurationException(format("`encryptedFields` is not configured for the collection %s.", namespace));
        }
        CodecRegistry codecRegistry = options.getKeyVaultMongoClientSettings() == null
                ? MongoClientSettings.getDefaultCodecRegistry()
                : options.getKeyVaultMongoClientSettings().getCodecRegistry();

        BsonDocument encryptedFields = rawEncryptedFields.toBsonDocument(BsonDocument.class, codecRegistry);
        BsonValue fields = encryptedFields.get("fields");
        if (fields != null) {
            if (!fields.isArray()) {
                throw new MongoConfigurationException(format("`encryptedFields` is incorrectly configured for the collection %s."
                        + " `encryptedFields.fields` must be an array, but is %s type.", namespace, fields.getBsonType()));
            }

            String kmsProvider = createEncryptedCollectionParams.getKmsProvider();
            DataKeyOptions dataKeyOptions = new DataKeyOptions();
            BsonDocument masterKey = createEncryptedCollectionParams.getMasterKey();
            if (masterKey != null) {
                dataKeyOptions.masterKey(masterKey);
            }
            String keyIdBsonKey = "keyId";

            return Flux.concat(fields.asArray()
                            .stream()
                            .map(field -> {
                                Mono<BsonValue> newField = Mono.fromSupplier(() -> field);
                                if (field.isDocument()) {
                                    BsonDocument fieldDocument = field.asDocument();
                                    if (fieldDocument.get(keyIdBsonKey, BsonBoolean.FALSE).equals(BsonNull.VALUE)) {
                                        newField = Mono.fromDirect(createDataKey(kmsProvider, dataKeyOptions))
                                                .map(dataKeyId -> (BsonValue) fieldDocument.append(keyIdBsonKey, dataKeyId))
                                                .contextWrite(ctx -> ctx.put("createdDataKey", "true"));
                                    }
                                }
                                return newField;
                            })
                            .collect(Collectors.toList()))
                    .collectList()
                    .map(fieldsArray -> encryptedFields.clone().append("fields", new BsonArray(fieldsArray)))
                    .flatMap(actualEncryptedFields ->
                            Mono.deferContextual(ctx ->
                                    Mono.from(database.createCollection(collectionName,
                                                    createCollectionOptions.clone().encryptedFields(actualEncryptedFields)))
                                            .doOnError(t -> ctx.hasKey("createdDataKey"), e -> {
                                                throw new MongoUpdatedEncryptedFieldsException(actualEncryptedFields,
                                                        format("Failed to create %s.", namespace), e);
                                            })
                                            .thenReturn(actualEncryptedFields))
                    );
        }
        return Mono.from(database.createCollection(collectionName, createCollectionOptions))
                .thenReturn(encryptedFields);
    }

User input validation is done outside of the Publisher.
All side effects are kept within the bounds of the Publishers and propagated downstream for consumption by later Publishers.
I had to use the Context-Propagation to capture if I had called createDataKey for "Case 3: Invalid keyId". Not used that before but it is handy.
This should remove the need for using the oneShotPublisher - although like 99% of our Publishers its nonsensical to subscribe multiple times.

...ync/src/test/functional/com/mongodb/client/AbstractClientSideEncryptionAutoDataKeysTest.java

stIncMale · 2023-02-09T01:15:27Z

@rozza Thanks! I had to read-up about Context.

All side effects are kept within the bounds of the Publishers and propagated downstream for consumption by later Publishers.
I had to use the Context-Propagation to capture if I had called createDataKey

Context can't be used to propagate information downstream because it only propagates information upstream (from a subscriber through the chain of operators to the source of the data). I read about this in the section Adding a Context to a Reactive Sequence.

I also experimented just in case, and as expected, the context entry with the createdDataKey key is not propagated downstream.

I additionally have doubts about this code:

...
.collectList()
.map(fieldsArray -> encryptedFields.clone().append("fields", new BsonArray(fieldsArray)))
.flatMap(actualEncryptedFields -> ...

My understanding is that the function provided to map is called only if collectList() emits a list, which happens only if there were no errors. On top of that, collectList is documented as "discards the elements in the List upon cancellation or error triggered by a data signal."

However, I realized that the publisher indeed does not have to be one-shot because the mutable state it captures is created once per Subscriber by the Supplier passed to Mono.defer.

I took your idea to use doOnError(Predicate, Consumer).

- treat errors while encoding `rawEncryptedFields` as fatal, i.e., do that outside of the publisher pipeline; - remove the one-shot publisher as it's not needed; - simplify the `doOnError` call; - undo the change proposed by DRIVERS-2540. JAVA-4679

rozza

LGTM!

I had a couple of thoughts regarding your comments - but they don't require any actioning:

Context can't be used to propagate information downstream because it only propagates information upstream

According to the context.write docs:

The Context is propagated from the bottom of the chain towards the top.

So I read that as in order of the chain (operations), the bottom being the publishersOfUpdatedFields in your code and the top being the final Mono.

This makes sense to me as, I used the context in order to get the invalidKeyIdtest to pass which was failing when not using the context.

My understanding is that the function provided to map is called only if collectList() emits a list, which happens only if there were no errors. On top of that, collectList is documented as "discards the elements in the List upon cancellation or error triggered by a data signal."

One to check, if the fields list is empty, then collectList will either emit an empty list or just complete. The map stage becomes a noop either way and will complete without any changes.

The impact of an error or cancellation signals would be the same for either approach. An error in any stage will bubble up to the top and be caught by any error handlers. I think the same for cancellation signals as well.

stIncMale · 2023-02-09T15:24:20Z

@rozza, for the sake of knowledge sharing, I have to disagree:

The Context is propagated from the bottom of the chain towards the top.
So I read that as in order of the chain (operations), the bottom being the publishersOfUpdatedFields in your code and the top being the final Mono.

The following example from the docs:

String key = "message";
Mono<String> r = Mono.just("Hello")
    .contextWrite(ctx -> ctx.put(key, "World")) 
    .flatMap( s -> Mono.deferContextual(ctx ->
        Mono.just(s + " " + ctx.getOrDefault(key, "Stranger")))); 

StepVerifier.create(r)
            .expectNext("Hello Stranger") 
            .verifyComplete();

demonstrates what "from the bottom of the chain towards the top" mean. If it worked according to your interpretation, the emitted string would have been "Hello World", but it is "Hello Stranger".

I used the context in order to get the invalidKeyIdtest to pass which was failing when not using the context.

The invalidKeyId test neither triggers nor checks for MongoUpdatedEncryptedFieldsException, so even if we remove the doOnError call from either your or my implementation, it passes.

One to check, if the fields list is empty, then collectList will either emit an empty list or just complete. The map stage becomes a noop either way and will complete without any changes.

The impact of an error or cancellation signals would be the same for either approach. An error in any stage will bubble up to the top and be caught by any error handlers. I think the same for cancellation signals as well.

As I mentioned in the PR description, not all needed tests are done in the spec, the spec change in DRIVERS-2538 is still not done. However, we can experiment by temporarily changing the code outside of the createEncryptedCollection method.

If we modify ClientEncryptionImpl.createDataKey such that it throws an exception, createEncryptedCollection must emit MongoUpdatedEncryptedFieldsException. The implementation proposed in Add the ClientEncryption.createEncryptedCollection helper method #1079 (comment) does not do that, while the implementation in the PR does.
If we modify ClientEncryptionImpl.createDataKey such that it throws an exception the second time it's called, and modify the insertEncryptedValue such that it has two elements in fields with keyId: null, then not only createEncryptedCollection must emit MongoUpdatedEncryptedFieldsException, but also MongoUpdatedEncryptedFieldsException.getEncryptedFields must report the key ID of the key that was created successfully. Again, the proposed implementation does not do that. Moreover, the function passed into flatMap is never called in this scenario, i.e., ctx never comes into play.

I made these modifications in the JAVA-4679_experiments branch (commit stIncMale@8ba7938) in case you want to try yourself.

Add the ClientEncryption.createEncryptedCollection helper method

0714f57

JAVA-4679

stIncMale force-pushed the JAVA-4679 branch from 3251a35 to 0714f57 Compare February 4, 2023 01:17

stIncMale commented Feb 4, 2023

View reviewed changes

driver-core/src/main/com/mongodb/client/model/CreateCollectionOptions.java Outdated Show resolved Hide resolved

stIncMale commented Feb 4, 2023

View reviewed changes

stIncMale requested review from rozza and katcharov February 4, 2023 02:13

stIncMale commented Feb 4, 2023

View reviewed changes

stIncMale added 2 commits February 3, 2023 22:58

Simplify the reactive implementation a bit

4963de7

JAVA-4679

Remove the KMS provider name duplication in the test

ac41300

JAVA-4679

rozza reviewed Feb 6, 2023

View reviewed changes

katcharov reviewed Feb 7, 2023

View reviewed changes

Address review concerns

0016d2c

JAVA-4679

stIncMale requested review from rozza and katcharov February 7, 2023 01:33

Update a comment

fd6eebe

JAVA-4679

katcharov approved these changes Feb 7, 2023

View reviewed changes

rozza reviewed Feb 7, 2023

View reviewed changes

...ync/src/test/functional/com/mongodb/client/AbstractClientSideEncryptionAutoDataKeysTest.java Show resolved Hide resolved

stIncMale mentioned this pull request Feb 7, 2023

Introduce BsonUtil.mutableDeepCopy #1081

Merged

Do some changes:

6dd7746

- treat errors while encoding `rawEncryptedFields` as fatal, i.e., do that outside of the publisher pipeline; - remove the one-shot publisher as it's not needed; - simplify the `doOnError` call; - undo the change proposed by DRIVERS-2540. JAVA-4679

stIncMale requested a review from rozza February 9, 2023 01:28

rozza approved these changes Feb 9, 2023

View reviewed changes

Replace doOnError with onErrorMap

3cc7102

rozza approved these changes Feb 9, 2023

View reviewed changes

stIncMale merged commit 59585de into mongodb:master Feb 9, 2023

stIncMale deleted the JAVA-4679 branch February 9, 2023 16:22

stIncMale self-assigned this Feb 19, 2023

Add the ClientEncryption.createEncryptedCollection helper method #1079

Add the ClientEncryption.createEncryptedCollection helper method #1079

Uh oh!

Conversation

stIncMale commented Feb 4, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Uh oh!

stIncMale Feb 4, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

stIncMale Feb 4, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

stIncMale Feb 4, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

stIncMale Feb 9, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

rozza left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

stIncMale commented Feb 7, 2023

Uh oh!

katcharov left a comment

Choose a reason for hiding this comment

Uh oh!

rozza commented Feb 7, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Uh oh!

stIncMale commented Feb 9, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

rozza left a comment

Choose a reason for hiding this comment

Uh oh!

stIncMale commented Feb 9, 2023

Uh oh!

Uh oh!

Add the `ClientEncryption.createEncryptedCollection` helper method #1079

Add the `ClientEncryption.createEncryptedCollection` helper method #1079

stIncMale commented Feb 4, 2023 •

edited

Loading

stIncMale Feb 4, 2023 •

edited

Loading

stIncMale Feb 4, 2023 •

edited

Loading

stIncMale Feb 4, 2023 •

edited

Loading

stIncMale Feb 9, 2023 •

edited

Loading

rozza commented Feb 7, 2023 •

edited

Loading

stIncMale commented Feb 9, 2023 •

edited

Loading