Sync streams

[simolus3]

This ports sync streams (Dart PR: powersync-ja/powersync.dart#317) to the JavaScript SDK.

Subscriptions are requested by calling db.syncStream('stream_name', params).subscribe(). The subscribe() call returns a subscription handle which can be used to unsubscribe() the stream. As an additional measure to ensure streams don't leak, we also add a finalization registry on streams to call unsubscribe() implicitly.

When a stream is subscribed to while offline, we don't really have to do much in the SDK. Most of the logic is implemented in the core extension, which will remember the subscription (so that, if we're connecting later but within the TTL, the stream is included even if the subscription no longer exists).
Because all streams (even those that have never been resolved - relevant for introspection) are included in the sync status, we need to update the sync status for new subscriptions even while offline. The new powersync_offline_sync_status() funciton can be used for that, it also replaces the query on ps_sync_state used during initialization.

To report the current status of stream subscriptions, we can simply take them from the JSON provided by the core extension.

While connected, the core extension needs to know all streams that have a subscription active. This information is provided in two ways:

1. When connect() is called, we provide a snapshot of all streams that are currently active.
2. When that list changes while connected, we call updateSubscriptions() with the new set of stream subscriptions. The core extension is informed about this change, which has two effects:
   • When subscriptions are removed, their expiry date is no longer periodically updated.
   • When a subscription is added, or if the expiry date of an inactive subscription expires, the core extension requests the sync iteration to restart. The next iteration would then request the new actual set of stream subscriptions.

Within a PowerSyncDatabase instance, it is possible to subscribe to the same stream / params multiple times. The subscriptions we pass to the core extension is a de-duplicated set of all active subscription instances (managed with a refcount).

With a shared sync worker, the same concept applies across different tabs as well: Each tabs sends its de-duplicated set of active subscriptions to the worker, which then applies its own de-duplication logic before notifying the core extension.
The shared worker also keeps track of which tabs are owning which subcriptions. That allows it to update the set when a tab is closing (in case some subscriptions were only active in a single tab).

TODO:

• (probably best in a separate PR): Use the weblocks hack to detect closing tabs from the shared worker.
• Test different subscriptions across tabs

I will add hooks for subscriptions in a follow-up PR.

[changeset-bot]

⚠️ No Changeset found

Latest commit: c9b757a

Merging this PR will not cause a version bump for any packages. If these changes should not result in a new version, you're good to go. If these changes should result in a version bump, you need to add a changeset.

This PR includes no changesets

When changesets are added to this PR, you'll see the packages that this PR includes changesets for and the associated semver types

Click here to learn what changesets are, and how to add one.

Click here if you're a maintainer who wants to add a changeset to this PR

[rkistner]

I still need to review properly, but this part stood out to me:

As an additional measure to ensure streams don't leak, we also add a finalization registry on streams to call unsubscribe() implicitly.

I wonder if this could break some common use cases? For example, a developer may just want to subscribe to relevant streams when loading a page:

async function loadData(id) {
  const subscription = db.syncStream('my_stream', {id}).subscribe();
  await subscription.waitForFirstSync();
}

However, the subscription is garbage collected, which would trigger the implicit unsubscribe logic.

Of course, we don't do that, and the user uses a long-running SPA, the subscription could stay alive forever, long after the user navigated to a different page.

So it feels like overall for an SPA, the developer will need to keep track of these subscriptions somewhere. I'm just not sure what the best behavior is if they don't - both unsubscribing automatically and not doing that has potential for issues. Maybe instead just logging a warning when a subscription is garbage collected without an explicit unsubscribe? We'll also need some guides on how to handle this with some common frameworks.

[simolus3]

Maybe instead just logging a warning when a subscription is garbage collected without an explicit unsubscribe?

That sounds good to me, I've changed the finalizer to a warning.

We'll also need some guides on how to handle this with some common frameworks.

I assume that at least in React, we'd have a useSyncStream instead of manual subscribe() and unsubscribe() calls.

[stevensJourney]

I'm quite happy with the changes from my side. I could not spot any concerns or potential issues. This looks good to me :)

[stevensJourney]

This is very nice