Optimize incremental sync: Phase 1

.changeset/chilled-mirrors-marry.md

@@ -0,0 +1,13 @@
+---
+'@powersync/service-module-postgres-storage': minor
+'@powersync/service-module-mongodb-storage': minor
+'@powersync/service-core-tests': minor
+'@powersync/service-module-postgres': minor
+'@powersync/service-module-mongodb': minor
+'@powersync/service-core': minor
+'@powersync/service-module-mysql': minor
+'@powersync/service-sync-rules': minor
+'@powersync/service-errors': patch
+---
+
+Minor optimizations to new checkpoint calulations.

libs/lib-postgres/src/db/connection/AbstractPostgresConnection.ts

@@ -7,9 +7,7 @@ export type DecodedSQLQueryExecutor<T extends t.Codec<any, any>> = {
   rows: () => Promise<t.Decoded<T>[]>;
 };
 
-export abstract class AbstractPostgresConnection<
-  Listener extends framework.DisposableListener = framework.DisposableListener
-> extends framework.DisposableObserver<Listener> {
+export abstract class AbstractPostgresConnection<Listener = {}> extends framework.BaseObserver<Listener> {
   protected abstract baseConnection: pgwire.PgClient;
 
   stream(...args: pgwire.Statement[]): AsyncIterableIterator<pgwire.PgChunk> {

libs/lib-postgres/src/db/connection/ConnectionSlot.ts

@@ -1,7 +1,7 @@
 import * as framework from '@powersync/lib-services-framework';
 import * as pgwire from '@powersync/service-jpgwire';
 
-export interface NotificationListener extends framework.DisposableListener {
+export interface NotificationListener {
   notification?: (payload: pgwire.PgNotification) => void;
 }
 
@@ -23,7 +23,7 @@ export type ConnectionSlotOptions = {
 
 export const MAX_CONNECTION_ATTEMPTS = 5;
 
-export class ConnectionSlot extends framework.DisposableObserver<ConnectionSlotListener> {
+export class ConnectionSlot extends framework.BaseObserver<ConnectionSlotListener> {
   isAvailable: boolean;
   isPoking: boolean;
 
@@ -63,7 +63,7 @@ export class ConnectionSlot extends framework.DisposableObserver<ConnectionSlotL
     this.closed = true;
     const connection = this.connection ?? (await this.connectingPromise);
     await connection?.end();
-    return super[Symbol.dispose]();
+    super.clearListeners();
   }
 
   protected async configureConnectionNotifications(connection: pgwire.PgConnection) {

libs/lib-services/src/utils/BaseObserver.ts

@@ -30,4 +30,8 @@ export class BaseObserver<T> implements ObserverClient<T> {
       await cb(this.listeners[i]);
     }
   }
+
+  clearListeners() {
+    this.listeners = {};
+  }
 }

libs/lib-services/src/utils/utils-index.ts

@@ -1,3 +1,2 @@
 export * from './BaseObserver.js';
-export * from './DisposableObserver.js';
 export * from './environment-variables.js';