Cleanup

Author: simolus3

packages/common/src/client/sync/bucket/SqliteBucketStorage.ts

@@ -416,16 +416,6 @@ export class SqliteBucketStorage extends BaseObserver<BucketStorageListener> imp
 
   async control(op: string, payload: string | ArrayBuffer | null): Promise<string> {
     return await this.writeTransaction(async (tx) => {
-      if (payload == null || typeof payload == 'string') {
-        console.log('starting op', op, payload);
-      } else {
-        console.log(
-          'starting binary op',
-          op,
-          Array.prototype.map.call(new Uint8Array(payload), (x) => x.toString(16).padStart(2, '0')).join('')
-        );
-      }
-
       const [[raw]] = await tx.executeRaw('SELECT powersync_control(?, ?)', [op, payload]);
       return raw;
     });

packages/common/src/client/sync/stream/AbstractRemote.ts

@@ -20,8 +20,6 @@ export type RemoteConnector = {
 const POWERSYNC_TRAILING_SLASH_MATCH = /\/+$/;
 const POWERSYNC_JS_VERSION = PACKAGE.version;
 
-// Refresh at least 30 sec before it expires
-const REFRESH_CREDENTIALS_SAFETY_PERIOD_MS = 30_000;
 const SYNC_QUEUE_REQUEST_LOW_WATER = 5;
 
 // Keep alive message is sent every period
@@ -340,7 +338,7 @@ export abstract class AbstractRemote {
     const stream = new DataStream({
       logger: this.logger,
       pressure: {
-        lowWaterMark: Math.max(1, Math.round(syncQueueRequestSize * 0.7))
+        lowWaterMark: SYNC_QUEUE_REQUEST_LOW_WATER
       }
     });
 

packages/common/src/client/sync/stream/AbstractStreamingSyncImplementation.ts

@@ -837,6 +837,10 @@ The next upload iteration will be delayed.`);
 
     const abortController = new AbortController();
     signal.addEventListener('abort', () => abortController.abort());
+
+    // Pending sync lines received from the service, as well as local events that trigger a powersync_control
+    // invocation (local events include refreshed tokens and completed uploads).
+    // This is a single data stream so that we can handle all control calls from a single place.
     let controlInvocations: DataStream<[string, ArrayBuffer | string | undefined]> | null = null;
 
     async function connect(instr: EstablishSyncStream) {
@@ -854,8 +858,7 @@ The next upload iteration will be delayed.`);
             ...syncOptions,
             fetchStrategy: resolvedOptions.fetchStrategy
           },
-          // TODO: Can we avoid the copy here?
-          (buffer) => ['line_binary', new Uint8Array(buffer).buffer]
+          (buffer) => ['line_binary', buffer]
         );
       }