Scan bounded ranges.

Author: rkistner

modules/module-slatedb-storage/src/storage/SlateDBKVStore.ts

@@ -156,7 +156,10 @@ export class SlateDBKVStore implements AsyncDisposable {
     }
   }
 
-  scanPrefix<T = unknown>(prefix: SlateDBKey, options?: { limit?: number }): AsyncIterable<SlateDBEntry<T>> {
+  scanPrefix<T = unknown>(
+    prefix: SlateDBKey,
+    options?: { limit?: number; startAfter?: SlateDBKey; endAt?: SlateDBKey }
+  ): AsyncIterable<SlateDBEntry<T>> {
     return scanPrefixFrom<T>(this.db, prefix, options);
   }
 
@@ -269,7 +272,10 @@ export class SlateDBCheckpointReader {
     return getFrom<T>(this.reader, key);
   }
 
-  scanPrefix<T = unknown>(prefix: SlateDBKey, options?: { limit?: number }): AsyncIterable<SlateDBEntry<T>> {
+  scanPrefix<T = unknown>(
+    prefix: SlateDBKey,
+    options?: { limit?: number; startAfter?: SlateDBKey; endAt?: SlateDBKey }
+  ): AsyncIterable<SlateDBEntry<T>> {
     return scanPrefixFrom<T>(this.reader, prefix, options);
   }
 }
@@ -306,9 +312,9 @@ async function getFrom<T>(reader: SlateDBReadable, key: SlateDBKey): Promise<T |
 async function* scanPrefixFrom<T>(
   reader: SlateDBReadable,
   prefix: SlateDBKey,
-  options: { limit?: number } = {}
+  options: { limit?: number; startAfter?: SlateDBKey; endAt?: SlateDBKey } = {}
 ): AsyncIterable<SlateDBEntry<T>> {
-  const iterator = await reader.scan_prefix(toKeyBytes(prefix), unboundedRange());
+  const iterator = await reader.scan_prefix(toKeyBytes(prefix), scanRange(options));
   try {
     let count = 0;
     while (options.limit == null || count < options.limit) {
@@ -332,12 +338,12 @@ function toKeyBytes(key: SlateDBKey): Uint8Array {
   return typeof key == 'string' ? textEncoder.encode(key) : key;
 }
 
-function unboundedRange(): KeyRange {
+function scanRange(options: { startAfter?: SlateDBKey; endAt?: SlateDBKey }): KeyRange {
   return {
-    start: undefined,
+    start: options.startAfter == null ? undefined : toKeyBytes(options.startAfter),
     start_inclusive: false,
-    end: undefined,
-    end_inclusive: false
+    end: options.endAt == null ? undefined : toKeyBytes(options.endAt),
+    end_inclusive: true
   };
 }
 

modules/module-slatedb-storage/src/storage/SlateDBSyncBucketStorage.ts

@@ -220,6 +220,9 @@ export class SlateDBSyncBucketStorage
     for (const request of dataBuckets) {
       let start = request.start;
       while (emitted < limit) {
+        if (start >= checkpoint.checkpoint) {
+          break;
+        }
         const chunk: utils.SyncBucketData = {
           bucket: request.bucket,
           after: utils.internalToExternalOpId(start),
@@ -230,7 +233,8 @@ export class SlateDBSyncBucketStorage
         let chunkSizeBytes = 0;
 
         for await (const entry of this.store.scanPrefix<BucketOpRecord>(
-          storagePrefix('bucket-data', this.replicationStreamId, request.bucket)
+          storagePrefix('bucket-data', this.replicationStreamId, request.bucket),
+          { startAfter: encodeOpId(start), endAt: encodeOpId(checkpoint.checkpoint) }
         )) {
           const op = BigInt(entry.value.op_id_bigint);
           if (op <= start || op > checkpoint.checkpoint) {