feat: Allow optional seq for end-of-sequence marker

Signed-off-by: Ryan Roberts <ryan@blockchaintp.com>

src/main/java/com/blockchaintp/daml/stores/qldb/QldbTransactionLog.java

@@ -17,6 +17,7 @@
 import java.util.Arrays;
 import java.util.Comparator;
 import java.util.Map;
+import java.util.Optional;
 import java.util.UUID;
 import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
@@ -242,9 +243,10 @@ private Tuple2<Long, Map.Entry<UUID, ByteString>> fromResult(final IonValue resu
   }
 
   @Override
-  public Observable<Map.Entry<UUID, ByteString>> from(final Long offset) {
+  public Observable<Map.Entry<UUID, ByteString>> from(final Optional<Long> offset) {
     tables.checkTables();
-    var readSeq = new QldbTxSeq(offset);
+
+    final var readSeq = getQldbTxSeq(offset);
     var queryPattern = "select s.%s,d.%s,d.%s from %s as d BY d_id, %s as s where d_id = s.%s and s.%s in ( %s )";
     return Observable.interval(pollInterval, TimeUnit.MILLISECONDS).map(x -> readSeq.peekRange(pageSize))
         .flatMap(seq -> driver.execute(tx -> {
@@ -263,4 +265,14 @@ public Observable<Map.Entry<UUID, ByteString>> from(final Long offset) {
         }));
   }
 
+  private QldbTxSeq getQldbTxSeq(final Optional<Long> offset) {
+    QldbTxSeq readSeq;
+    if (offset.isEmpty()) {
+      readSeq = new QldbTxSeq(this.seqSource.peekNext());
+    } else {
+      readSeq = new QldbTxSeq(offset.get());
+    }
+    return readSeq;
+  }
+
 }

src/main/java/com/blockchaintp/daml/stores/qldb/QldbTxSeq.java

@@ -23,6 +23,9 @@
  * A long sequence, initialisable from a QLDB tx log sequence table or an explicit point.
  */
 public final class QldbTxSeq implements SeqSource<Long> {
+  /**
+   * Marker for the start of a sequence.
+   */
   private long current;
 
   /**

src/main/java/com/blockchaintp/daml/stores/service/TransactionLogReader.java

@@ -14,6 +14,7 @@
 package com.blockchaintp.daml.stores.service;
 
 import java.util.Map;
+import java.util.Optional;
 
 import io.reactivex.rxjava3.core.Observable;
 
@@ -31,7 +32,8 @@ public interface TransactionLogReader<I, K, V> {
    * Stream committed log entries starting at offset.
    *
    * @param offset
+   *          - use None to start at the last commit
    * @return A stream of comitted log entires.
    */
-  Observable<Map.Entry<K, V>> from(I offset);
+  Observable<Map.Entry<K, V>> from(Optional<I> offset);
 }

src/test/java/com/blockchaintp/daml/stores/qldb/QldbTransactionLogIntegrationTest.java

@@ -29,6 +29,7 @@
 import software.amazon.qldb.QldbDriver;
 
 import java.util.ArrayList;
+import java.util.Optional;
 import java.util.UUID;
 import java.util.stream.Collectors;
 
@@ -79,7 +80,7 @@ final void committed_transactions_are_read_in_commit_order() throws StoreWriteEx
       txLog.commit(id);
     }
 
-    var stream = txLog.from(0L);
+    var stream = txLog.from(Optional.of(0L));
 
     Assertions.assertIterableEquals(ids,
         stream.take(30).map(x -> x.getKey()).collect(Collectors.toList()).blockingGet());