title: Avoid MongoDB read skew in reactive transactions with Spring?
body: <p>I'm using MongoDB 4.2 with Spring Boot 2.3.1 and I'm looking for a way to avoid read skew in my scenario.</p>
  <p>I have a collection named &quot;metadata&quot; and one named &quot;messages&quot;. The latter contains messages like this:</p>
  <pre><code>{
    &quot;aggregateId&quot; : &quot;myAggregateId&quot;,
    &quot;messageId&quot; : &quot;uuid&quot;, 
    &quot;message&quot; : &quot;some message&quot;
  }
  </code></pre>
  <p>and &quot;metadata&quot; contains the version for each &quot;aggregate&quot;:</p>
  <pre><code>{
    &quot;aggregateId&quot; : &quot;myAggregateId&quot;,
    &quot;version&quot; : NumberLong(40)
  }
  </code></pre>
  <p>The reason for not just storing messages in a subarray is among other things that the number of messages per aggregate can be greater than 16Mb (which is the document limit in MongoDB).</p>
  <p>When issuing a query I <em>think</em> I'd like to create an interface like this for the users:</p>
  <pre class="lang-java prettyprint-override"><code>public interface MyRepository {
       Mono&lt;Aggregate&gt; findByAggregateId(String aggregateId);
  }
  </code></pre>
  <p>where <code>Aggregate</code> is defined like this:</p>
  <pre><code>public class Aggregate {
      private final String aggregateId;
      private final int version;
      private Flux&lt;Message&gt; messages;
  }
  </code></pre>
  <p>The problem now is that I'd like <code>Aggregate</code> to be consistent when reading! I.e. if there are writes to the same aggregate <em>before</em> <code>messages</code> are subscribed to then I <em>don't</em> want the new messages to be included (those written after I've subscribed to <code>Mono&lt;Aggregate&gt;</code>).</p>
  <p>Let's look at an example. This is one attempt at an implementation:</p>
  <pre><code>public Mono&lt;Aggregate&gt; findByAggregateId(String aggregateId) {
      return transactionalOperator.execute(status -&gt; 
          reactiveMongoTemplate.findOne(query(where(&quot;aggregateId&quot;).is(aggregateId)), Document.class, &quot;metadata&quot;)
                               .map(metadata -&gt; {
                                  Aggregate aggregate = new Aggregate(metadata.getString(&quot;aggregateId&quot;), metadata.getLong(&quot;version&quot;));
                                  Flux&lt;Message&gt; messages = reactiveMongoTemplate.find(query, Message.class, &quot;messages&quot;);
                                  aggregate.setMessages(messages);
                                  return aggregate;
                               })
      );
  }
  </code></pre>
  <p>I totally understand that this won't work since the <code>messages</code> <code>Flux</code> is not subscribed to in the transaction. But I can't figure out how I should combine the outer <code>Aggregate</code> that is a <code>Mono</code> and an inner <code>Flux</code> (messages) and retain the non-blocking capabilities AND consistency (i.e. avoid read skew)?</p>
  <p>One approach would be to change the <code>Aggregate</code> class to this:</p>
  <pre><code>public class Aggregate {
      private final String aggregateId;
      private final int version;
      private Stream&lt;Message&gt; messages;
  }
  </code></pre>
  <p>and change the <code>findByAggregateId</code> implementation to this:</p>
  <pre><code>public Mono&lt;Aggregate&gt; findByAggregateId(String aggregateId) {
      return transactionalOperator.execute(status -&gt; 
          reactiveMongoTemplate.findOne(query(where(&quot;aggregateId&quot;).is(aggregateId)), Document.class, &quot;metadata&quot;)
                               .flatMap(metadata -&gt; {
                                  Aggregate aggregate = new Aggregate(metadata.getString(&quot;aggregateId&quot;), metadata.getLong(&quot;version&quot;));
                                  Stream&lt;Message&gt; messages = reactiveMongoTemplate.find(query, Message.class, &quot;messages&quot;).toStream();
                                  aggregate.setMessages(messages);
                                  return aggregate;
                               })
      );
  }
  </code></pre>
  <p>but calling <code>toStream</code> is a blocking operation so this is not right.</p>
  <p>So what is the correct way to deal with this?</p>