Unit tests for bucket checksum state

Author: simolus3

packages/service-core/src/routes/endpoints/socket-route.ts

@@ -58,8 +58,6 @@ export const syncStreamReactive: SocketRouteGenerator = (router) =>
         return;
       }
 
-      const syncParams = new RequestParameters(context.token_payload!, params.parameters ?? {});
-
       const {
         storageEngine: { activeBucketStorage }
       } = service_context;
@@ -95,7 +93,6 @@ export const syncStreamReactive: SocketRouteGenerator = (router) =>
             ...params,
             binary_data: true // always true for web sockets
           },
-          syncParams,
           token: context!.token_payload!,
           tokenStreamOptions: {
             // RSocket handles keepalive events by default

packages/service-core/src/routes/endpoints/sync-stream.ts

@@ -52,9 +52,6 @@ export const syncStreamed = routeDefinition({
       });
     }
 
-    const params: util.StreamingSyncRequest = payload.params;
-    const syncParams = new RequestParameters(payload.context.token_payload!, payload.params.parameters ?? {});
-
     const bucketStorage = await storageEngine.activeBucketStorage.getActiveStorage();
 
     if (bucketStorage == null) {
@@ -75,8 +72,7 @@ export const syncStreamed = routeDefinition({
         syncContext: syncContext,
         bucketStorage,
         syncRules: syncRules,
-        params,
-        syncParams,
+        params: payload.params,
         token: payload.context.token_payload!,
         tracker,
         signal: controller.signal,

packages/service-core/src/sync/BucketChecksumState.ts

@@ -4,6 +4,7 @@ import {
   BucketSource,
   BucketSourceType,
   RequestedStream,
+  RequestJwtPayload,
   RequestParameters,
   ResolvedBucket,
   SqlSyncRules
@@ -28,10 +29,9 @@ export interface BucketChecksumStateOptions {
   syncContext: SyncContext;
   bucketStorage: BucketChecksumStateStorage;
   syncRules: SqlSyncRules;
-  syncParams: RequestParameters;
+  tokenPayload: RequestJwtPayload;
   syncRequest: util.StreamingSyncRequest;
   logger?: Logger;
-  initialBucketPositions?: { name: string; after: util.InternalOpId }[];
 }
 
 type BucketSyncState = {
@@ -79,14 +79,14 @@ export class BucketChecksumState {
       options.syncContext,
       options.bucketStorage,
       options.syncRules,
-      options.syncParams,
+      options.tokenPayload,
       options.syncRequest,
       this.logger
     );
     this.bucketDataPositions = new Map();
 
-    for (let { name, after: start } of options.initialBucketPositions ?? []) {
-      this.bucketDataPositions.set(name, { start_op_id: start });
+    for (let { name, after: start } of options.syncRequest.buckets ?? []) {
+      this.bucketDataPositions.set(name, { start_op_id: BigInt(start) });
     }
   }
 
@@ -199,7 +199,7 @@ export class BucketChecksumState {
       }));
       bucketsToFetch = [...generateBucketsToFetch].map((b) => {
         return {
-          ...bucketDescriptionMap.get(b)!,
+          priority: bucketDescriptionMap.get(b)!.priority,
           bucket: b
         };
       });
@@ -233,11 +233,11 @@ export class BucketChecksumState {
         message += `buckets: ${allBuckets.length} ${limitedBuckets(allBuckets, 20)}`;
         this.logger.info(message, { checkpoint: base.checkpoint, user_id: user_id, buckets: allBuckets.length });
       };
-      bucketsToFetch = allBuckets;
+      bucketsToFetch = allBuckets.map((b) => ({ bucket: b.bucket, priority: b.priority }));
 
       const subscriptions: util.StreamDescription[] = [];
       for (const source of this.parameterState.syncRules.bucketSources) {
-        if (source.type == BucketSourceType.SYNC_STREAM && this.parameterState.isSubscribedToStream(source)) {
+        if (this.parameterState.isSubscribedToStream(source)) {
           subscriptions.push({
             name: source.name,
             is_default: source.subscribedToByDefault
@@ -360,7 +360,11 @@ export class BucketParameterState {
   public readonly syncRules: SqlSyncRules;
   public readonly syncParams: RequestParameters;
   private readonly querier: BucketParameterQuerier;
-  private readonly staticBuckets: Map<string, BucketDescription>;
+  /**
+   * Static buckets. This map is guaranteed not to change during a request, since resolving static buckets can only
+   * take request parameters into account,
+   */
+  private readonly staticBuckets: Map<string, ResolvedBucket>;
   private readonly includeDefaultStreams: boolean;
   // Indexed by the client-side id
   private readonly explicitStreamSubscriptions: Record<string, util.RequestedStreamSubscription>;
@@ -375,22 +379,22 @@ export class BucketParameterState {
     context: SyncContext,
     bucketStorage: BucketChecksumStateStorage,
     syncRules: SqlSyncRules,
-    syncParams: RequestParameters,
+    tokenPayload: RequestJwtPayload,
     request: util.StreamingSyncRequest,
     logger: Logger
   ) {
     this.context = context;
     this.bucketStorage = bucketStorage;
     this.syncRules = syncRules;
-    this.syncParams = syncParams;
+    this.syncParams = new RequestParameters(tokenPayload, request.parameters ?? {});
     this.logger = logger;
 
     const idToStreamSubscription: Record<string, util.RequestedStreamSubscription> = {};
     const streamsByName: Record<string, RequestedStream[]> = {};
-    const subscriptions = request.subscriptions;
+    const subscriptions = request.streams;
     if (subscriptions) {
-      for (const subscription of subscriptions.opened) {
-        idToStreamSubscription[subscription.stream] = subscription;
+      for (const subscription of subscriptions.subscriptions) {
+        idToStreamSubscription[subscription.client_id] = subscription;
 
         const syncRuleStream: RequestedStream = {
           parameters: subscription.parameters ?? {},
@@ -412,7 +416,7 @@ export class BucketParameterState {
       streams: streamsByName
     });
 
-    this.staticBuckets = new Map<string, BucketDescription>(
+    this.staticBuckets = new Map<string, ResolvedBucket>(
       mergeBuckets(this.querier.staticBuckets).map((b) => [b.bucket, b])
     );
     this.lookups = new Set<string>(this.querier.parameterQueryLookups.map((l) => JSONBig.stringify(l.values)));
@@ -441,14 +445,13 @@ export class BucketParameterState {
     }
 
     return {
-      definition: description.definition,
       bucket: description.bucket,
       priority: priorityOverride ?? description.priority,
       subscriptions: description.inclusion_reasons.map((reason) => {
         if (reason == 'default') {
-          return { def: description.definition };
+          return { default: 0 }; // TODO
         } else {
-          return { sub: reason.subscription };
+          return reason.subscription;
         }
       })
     };
@@ -489,19 +492,19 @@ export class BucketParameterState {
    * For static buckets, we can keep track of which buckets have been updated.
    */
   private async getCheckpointUpdateStatic(checkpoint: storage.StorageCheckpointUpdate): Promise<CheckpointUpdate> {
-    const querier = this.querier;
+    const staticBuckets = [...this.staticBuckets.values()];
     const update = checkpoint.update;
 
     if (update.invalidateDataBuckets) {
       return {
-        buckets: querier.staticBuckets,
+        buckets: staticBuckets,
         updatedBuckets: INVALIDATE_ALL_BUCKETS
       };
     }
 
     const updatedBuckets = new Set<string>(getIntersection(this.staticBuckets, update.updatedDataBuckets));
     return {
-      buckets: querier.staticBuckets,
+      buckets: staticBuckets,
       updatedBuckets
     };
   }
@@ -512,7 +515,7 @@ export class BucketParameterState {
   private async getCheckpointUpdateDynamic(checkpoint: storage.StorageCheckpointUpdate): Promise<CheckpointUpdate> {
     const querier = this.querier;
     const storage = this.bucketStorage;
-    const staticBuckets = querier.staticBuckets;
+    const staticBuckets = this.staticBuckets.values();
     const update = checkpoint.update;
 
     let hasParameterChange = false;
@@ -556,7 +559,7 @@ export class BucketParameterState {
         }
       }
     }
-    const allBuckets = [...staticBuckets, ...dynamicBuckets];
+    const allBuckets = [...staticBuckets, ...mergeBuckets(dynamicBuckets)];
 
     if (invalidateDataBuckets) {
       return {
@@ -632,15 +635,15 @@ function limitedBuckets(buckets: string[] | { bucket: string }[], limit: number)
  * bucket.
  */
 function mergeBuckets(buckets: ResolvedBucket[]): ResolvedBucket[] {
-  const byDefinition: Record<string, ResolvedBucket> = {};
+  const byBucketId: Record<string, ResolvedBucket> = {};
 
   for (const bucket of buckets) {
-    if (Object.hasOwn(byDefinition, bucket.definition)) {
-      byDefinition[bucket.definition].inclusion_reasons.push(...bucket.inclusion_reasons);
+    if (Object.hasOwn(byBucketId, bucket.bucket)) {
+      byBucketId[bucket.bucket].inclusion_reasons.push(...bucket.inclusion_reasons);
     } else {
-      byDefinition[bucket.definition] = bucket;
+      byBucketId[bucket.bucket] = structuredClone(bucket);
     }
   }
 
-  return Object.values(byDefinition);
+  return Object.values(byBucketId);
 }

packages/service-core/src/sync/sync.ts

@@ -1,5 +1,11 @@
 import { JSONBig, JsonContainer } from '@powersync/service-jsonbig';
-import { BucketDescription, BucketPriority, RequestParameters, SqlSyncRules } from '@powersync/service-sync-rules';
+import {
+  BucketDescription,
+  BucketPriority,
+  RequestJwtPayload,
+  RequestParameters,
+  SqlSyncRules
+} from '@powersync/service-sync-rules';
 
 import { AbortError } from 'ix/aborterror.js';
 
@@ -19,7 +25,6 @@ export interface SyncStreamParameters {
   bucketStorage: storage.SyncRulesBucketStorage;
   syncRules: SqlSyncRules;
   params: util.StreamingSyncRequest;
-  syncParams: RequestParameters;
   token: auth.JwtPayload;
   logger?: Logger;
   /**
@@ -34,8 +39,7 @@ export interface SyncStreamParameters {
 export async function* streamResponse(
   options: SyncStreamParameters
 ): AsyncIterable<util.StreamingSyncLine | string | null> {
-  const { syncContext, bucketStorage, syncRules, params, syncParams, token, tokenStreamOptions, tracker, signal } =
-    options;
+  const { syncContext, bucketStorage, syncRules, params, token, tokenStreamOptions, tracker, signal } = options;
   const logger = options.logger ?? defaultLogger;
 
   // We also need to be able to abort, so we create our own controller.
@@ -58,7 +62,7 @@ export async function* streamResponse(
     bucketStorage,
     syncRules,
     params,
-    syncParams,
+    token,
     tracker,
     controller.signal,
     logger
@@ -86,25 +90,22 @@ async function* streamResponseInner(
   bucketStorage: storage.SyncRulesBucketStorage,
   syncRules: SqlSyncRules,
   params: util.StreamingSyncRequest,
-  syncParams: RequestParameters,
+  tokenPayload: RequestJwtPayload,
   tracker: RequestTracker,
   signal: AbortSignal,
   logger: Logger
 ): AsyncGenerator<util.StreamingSyncLine | string | null> {
   const { raw_data, binary_data } = params;
 
-  const checkpointUserId = util.checkpointUserId(syncParams.tokenParameters.user_id as string, params.client_id);
+  const userId = tokenPayload.sub;
+  const checkpointUserId = util.checkpointUserId(userId as string, params.client_id);
 
   const checksumState = new BucketChecksumState({
     syncContext,
     bucketStorage,
     syncRules,
-    syncParams,
+    tokenPayload,
     syncRequest: params,
-    initialBucketPositions: params.buckets?.map((bucket) => ({
-      name: bucket.name,
-      after: BigInt(bucket.after)
-    })),
     logger: logger
   });
   const stream = bucketStorage.watchCheckpointChanges({
@@ -229,7 +230,7 @@ async function* streamResponseInner(
           onRowsSent: markOperationsSent,
           abort_connection: signal,
           abort_batch: abortCheckpointSignal,
-          user_id: syncParams.userId,
+          user_id: userId,
           // Passing null here will emit a full sync complete message at the end. If we pass a priority, we'll emit a partial
           // sync complete message instead.
           forPriority: !isLast ? priority : null,

packages/service-core/src/util/protocol-types.ts

@@ -23,8 +23,6 @@ export const RequestedStreamSubscription = t.object({
   stream: t.string,
   /**
    * An opaque textual identifier assigned to this request by the client.
-   *
-   * Wh
    */
   client_id: t.string,
   /**
@@ -56,7 +54,7 @@ export const StreamSubscriptionRequest = t.object({
   /**
    * An array of sync streams the client has opened explicitly.
    */
-  opened: t.array(RequestedStreamSubscription)
+  subscriptions: t.array(RequestedStreamSubscription)
 });
 
 export type StreamSubscriptionRequest = t.Decoded<typeof StreamSubscriptionRequest>;
@@ -100,7 +98,7 @@ export const StreamingSyncRequest = t.object({
   /**
    * If the client is aware of streams, an array of streams the client has opened.
    */
-  subscriptions: StreamSubscriptionRequest.optional()
+  streams: StreamSubscriptionRequest.optional()
 });
 
 export type StreamingSyncRequest = t.Decoded<typeof StreamingSyncRequest>;
@@ -227,20 +225,32 @@ export type BucketSubscriptionReason = BucketDerivedFromDefaultStream | BucketDe
 
 /**
  * A bucket has been included in a checkpoint because it's part of a default stream.
- *
- * The string is the name of the stream definition.
  */
-export type BucketDerivedFromDefaultStream = { def: string };
+export type BucketDerivedFromDefaultStream = {
+  /**
+   * The index (into {@link Checkpoint.streams}) of the stream defining the bucket.
+   */
+  default: number;
+};
 
 /**
  * The bucket has been included in a checkpoint because it's part of a stream that a client has explicitly subscribed
  * to.
  *
  * The string is the client id associated with the subscription in {@link RequestedStreamSubscription}.
  */
-export type BucketDerivedFromExplicitSubscription = { sub: string };
+export type BucketDerivedFromExplicitSubscription = string;
 
-export interface ClientBucketDescription extends BucketDescription {
+export interface ClientBucketDescription {
+  /**
+   * An opaque id of the bucket.
+   */
+  bucket: string;
+  /**
+   * The priority used to synchronize this bucket, derived from its definition and an optional priority override from
+   * the stream subscription.
+   */
+  priority: BucketPriority;
   subscriptions: BucketSubscriptionReason[];
 }