Translate resolved buckets

Author: simolus3

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

@@ -1,4 +1,12 @@
-import { BucketDescription, isValidPriority, RequestParameters, SqlSyncRules } from '@powersync/service-sync-rules';
+import {
+  BucketDescription,
+  BucketPriority,
+  isValidPriority,
+  RequestedStream,
+  RequestParameters,
+  ResolvedBucket,
+  SqlSyncRules
+} from '@powersync/service-sync-rules';
 
 import * as storage from '../storage/storage-index.js';
 import * as util from '../util/util-index.js';
@@ -355,7 +363,9 @@ export class BucketParameterState {
   private readonly querier: BucketParameterQuerier;
   private readonly staticBuckets: Map<string, BucketDescription>;
   private readonly includeDefaultStreams: boolean;
-  private readonly explicitlyOpenedStreams: Record<string, util.OpenedStream>;
+  // Indexed by the client-side id
+  private readonly explicitStreamSubscriptions: Record<string, util.RequestedStreamSubscription>;
+  private readonly subscribedStreamNames: Set<string>;
   private readonly logger: Logger;
   private cachedDynamicBuckets: BucketDescription[] | null = null;
   private cachedDynamicBucketSet: Set<string> | null = null;
@@ -376,51 +386,73 @@ export class BucketParameterState {
     this.syncParams = syncParams;
     this.logger = logger;
 
-    const explicitlyOpenedStreams: Record<string, util.OpenedStream> = {};
+    const idToStreamSubscription: Record<string, util.RequestedStreamSubscription> = {};
+    const streamsByName: Record<string, RequestedStream[]> = {};
     const subscriptions = request.subscriptions;
     if (subscriptions) {
       for (const subscription of subscriptions.opened) {
-        explicitlyOpenedStreams[subscription.stream] = subscription;
+        idToStreamSubscription[subscription.stream] = subscription;
+
+        const syncRuleStream: RequestedStream = {
+          parameters: subscription.parameters ?? {},
+          opaque_id: subscription.client_id
+        };
+        if (Object.hasOwn(streamsByName, subscription.stream)) {
+          streamsByName[subscription.stream].push(syncRuleStream);
+        } else {
+          streamsByName[subscription.stream] = [syncRuleStream];
+        }
       }
     }
     this.includeDefaultStreams = subscriptions?.include_defaults ?? true;
-    this.explicitlyOpenedStreams = explicitlyOpenedStreams;
+    this.explicitStreamSubscriptions = idToStreamSubscription;
 
     this.querier = syncRules.getBucketParameterQuerier({
       globalParameters: this.syncParams,
       hasDefaultStreams: this.includeDefaultStreams,
-      resolveOpenedStream(name) {
-        const subscription = explicitlyOpenedStreams[name];
-        if (subscription) {
-          return subscription.parameters ?? {};
-        } else {
-          return null;
-        }
-      }
+      streams: streamsByName
     });
     this.staticBuckets = new Map<string, BucketDescription>(this.querier.staticBuckets.map((b) => [b.bucket, b]));
     this.lookups = new Set<string>(this.querier.parameterQueryLookups.map((l) => JSONBig.stringify(l.values)));
+    this.subscribedStreamNames = new Set(Object.keys(streamsByName));
   }
 
   /**
-   * Overrides the `description` based on subscriptions from the client.
-   *
-   * In partiuclar, this can override the priority assigned to a bucket.
+   * Translates an internal sync-rules {@link ResolvedBucket} instance to the public
+   * {@link util.ClientBucketDescription}.
    */
-  overrideBucketDescription(description: BucketDescription): BucketDescription {
-    const changedPriority = this.explicitlyOpenedStreams[description.definition]?.override_priority;
-    if (changedPriority != null && isValidPriority(changedPriority)) {
-      return {
-        ...description,
-        priority: changedPriority
-      };
-    } else {
-      return description;
+  translateResolvedBucket(description: ResolvedBucket): util.ClientBucketDescription {
+    // Assign
+    let priorityOverride: BucketPriority | null = null;
+    for (const reason of description.inclusion_reasons) {
+      if (reason != 'default') {
+        const requestedPriority = this.explicitStreamSubscriptions[reason.subscription]?.override_priority;
+        if (requestedPriority != null) {
+          if (priorityOverride == null) {
+            priorityOverride = requestedPriority as BucketPriority;
+          } else {
+            priorityOverride = Math.min(requestedPriority, priorityOverride) as BucketPriority;
+          }
+        }
+      }
     }
+
+    return {
+      definition: description.definition,
+      bucket: description.bucket,
+      priority: priorityOverride ?? description.priority,
+      subscriptions: description.inclusion_reasons.map((reason) => {
+        if (reason == 'default') {
+          return { def: description.definition };
+        } else {
+          return { sub: reason.subscription };
+        }
+      })
+    };
   }
 
   isSubscribedToStream(desc: SqlBucketDescriptor): boolean {
-    return (desc.subscribedToByDefault && this.includeDefaultStreams) || desc.name in this.explicitlyOpenedStreams;
+    return (desc.subscribedToByDefault && this.includeDefaultStreams) || this.subscribedStreamNames.has(desc.name);
   }
 
   async getCheckpointUpdate(checkpoint: storage.StorageCheckpointUpdate): Promise<CheckpointUpdate> {

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

@@ -16,11 +16,17 @@ export type BucketRequest = t.Decoded<typeof BucketRequest>;
 /**
  * A sync steam that a client has expressed interest in by explicitly opening it on the client side.
  */
-export const OpenedStream = t.object({
+export const RequestedStreamSubscription = t.object({
   /**
    * The defined name of the stream as it appears in sync stream definitions.
    */
   stream: t.string,
+  /**
+   * An opaque textual identifier assigned to this request by the client.
+   *
+   * Wh
+   */
+  client_id: t.string,
   /**
    * An optional dictionary of parameters to pass to this specific stream.
    */
@@ -34,26 +40,26 @@ export const OpenedStream = t.object({
   override_priority: t.number.optional()
 });
 
-export type OpenedStream = t.Decoded<typeof OpenedStream>;
+export type RequestedStreamSubscription = t.Decoded<typeof RequestedStreamSubscription>;
 
 /**
- * An overview of all opened streams as part of a streaming sync request.
+ * An overview of all subscribed streams as part of a streaming sync request.
  */
-export const OpenedStreams = t.object({
+export const StreamSubscriptionRequest = t.object({
   /**
    * Whether to sync default streams.
    *
-   * When disabled,only
+   * When disabled, only explicitly-opened subscriptions are included.
    */
   include_defaults: t.boolean.optional(),
 
   /**
    * An array of sync streams the client has opened explicitly.
    */
-  opened: t.array(OpenedStream)
+  opened: t.array(RequestedStreamSubscription)
 });
 
-export type StreamSubscriptions = t.Decoded<typeof OpenedStreams>;
+export type StreamSubscriptionRequest = t.Decoded<typeof StreamSubscriptionRequest>;
 
 export const StreamingSyncRequest = t.object({
   /**
@@ -94,7 +100,7 @@ export const StreamingSyncRequest = t.object({
   /**
    * If the client is aware of streams, an array of streams the client has opened.
    */
-  subscriptions: OpenedStreams.optional()
+  subscriptions: StreamSubscriptionRequest.optional()
 });
 
 export type StreamingSyncRequest = t.Decoded<typeof StreamingSyncRequest>;
@@ -107,7 +113,7 @@ export interface StreamingSyncCheckpointDiff {
   checkpoint_diff: {
     last_op_id: ProtocolOpId;
     write_checkpoint?: ProtocolOpId;
-    updated_buckets: BucketChecksumWithDescription[];
+    updated_buckets: CheckpointBucket[];
     removed_buckets: string[];
   };
 }
@@ -154,7 +160,7 @@ export interface StreamDescription {
 export interface Checkpoint {
   last_op_id: ProtocolOpId;
   write_checkpoint?: ProtocolOpId;
-  buckets: BucketChecksumWithDescription[];
+  buckets: CheckpointBucket[];
   streams: StreamDescription[];
 }
 
@@ -211,4 +217,31 @@ export interface BucketChecksum {
   count: number;
 }
 
-export interface BucketChecksumWithDescription extends BucketChecksum, BucketDescription {}
+/**
+ * The reason a particular bucket is included in a checkpoint.
+ *
+ * This information allows clients to associate individual buckets with sync streams they're subscribed to. Having that
+ * association is useful because it enables clients to track progress for individual sync streams.
+ */
+export type BucketSubscriptionReason = BucketDerivedFromDefaultStream | BucketDerivedFromExplicitSubscription;
+
+/**
+ * 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 };
+
+/**
+ * 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 interface ClientBucketDescription extends BucketDescription {
+  subscriptions: BucketSubscriptionReason[];
+}
+
+export interface CheckpointBucket extends BucketChecksum, ClientBucketDescription {}

packages/sync-rules/src/BucketDescription.ts

@@ -24,6 +24,7 @@ export interface BucketDescription {
    * The name of the sync rule or stream definition from which the bucket is derived.
    */
   definition: string;
+
   /**
    * The id of the bucket, which is derived from the name of the bucket's definition
    * in the sync rules as well as the values returned by the parameter queries.
@@ -34,3 +35,15 @@ export interface BucketDescription {
    */
   priority: BucketPriority;
 }
+
+/**
+ * A bucket that was resolved to a specific request including stream subscriptions.
+ *
+ * This includes information on why the bucket has been included in a checkpoint subset
+ * shown to clients.
+ */
+export interface ResolvedBucket extends BucketDescription {
+  inclusion_reasons: BucketInclusionReason[];
+}
+
+export type BucketInclusionReason = 'default' | { subscription: string };

packages/sync-rules/src/SqlSyncRules.ts

@@ -20,6 +20,7 @@ import {
   QueryParseOptions,
   RequestParameters,
   SourceSchema,
+  SqliteJsonRow,
   SqliteRow,
   StreamParseOptions,
   SyncRules
@@ -39,6 +40,21 @@ export interface SyncRulesOptions {
   throwOnError?: boolean;
 }
 
+export interface RequestedStream {
+  /**
+   * The parameters for the explicit stream subscription.
+   *
+   * Unlike {@link GetQuerierOptions.globalParameters}, these parameters are only applied to the particular stream.
+   */
+  parameters: SqliteJsonRow | null;
+
+  /**
+   * An opaque id of the stream subscription, used to associate buckets with the stream subscriptions that have caused
+   * them to be included.
+   */
+  opaque_id: string;
+}
+
 export interface GetQuerierOptions {
   globalParameters: RequestParameters;
   /**
@@ -48,13 +64,14 @@ export interface GetQuerierOptions {
    */
   hasDefaultStreams: boolean;
   /**
+   *
    * For streams, this is invoked to check whether the client has opened the relevant stream.
    *
    * @param name The name of the stream as it appears in the sync rule definitions.
    * @returns If the strema has been opened by the client, the stream parameters for that particular stream. Otherwise
    * null.
    */
-  resolveOpenedStream: (name: string) => Record<string, any> | null;
+  streams: Record<string, RequestedStream[]>;
 }
 
 export class SqlSyncRules implements SyncRules {
@@ -378,21 +395,33 @@ export class SqlSyncRules implements SyncRules {
       let params = options.globalParameters;
 
       if (descriptor.type == SqlBucketDescriptorType.STREAM) {
-        const opened = options.resolveOpenedStream(descriptor.name);
+        const subscriptions = options.streams[descriptor.name] ?? [];
 
-        if (!descriptor.subscribedToByDefault && opened == null) {
+        if (!descriptor.subscribedToByDefault && subscriptions.length) {
           // The client is not subscribing to this stream, so don't query buckets related to it.
           continue;
         }
 
-        if (opened != null) {
-          params = params.withAddedStreamParameters(opened);
+        let hasExplicitDefaultSubscription = false;
+        for (const subscription of subscriptions) {
+          let subscriptionParams = params;
+          if (subscription.parameters != null) {
+            subscriptionParams = params.withAddedStreamParameters(subscription.parameters);
+          } else {
+            hasExplicitDefaultSubscription = true;
+          }
+
+          queriers.push(descriptor.getBucketParameterQuerier(subscriptionParams));
         }
 
+        // If the stream is subscribed to by default and there is no explicit subscription that would match the default
+        // subscription, also include the default querier.
+        if (descriptor.subscribedToByDefault && !hasExplicitDefaultSubscription) {
+          queriers.push(descriptor.getBucketParameterQuerier(params));
+        }
+      } else {
         queriers.push(descriptor.getBucketParameterQuerier(params));
       }
-
-      queriers.push(descriptor.getBucketParameterQuerier(params));
     }
 
     return mergeBucketParameterQueriers(queriers);