close http streams better

Author: stevensJourney

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

@@ -8,7 +8,7 @@ import { DataStream } from '../../../utils/DataStream.js';
 import { PowerSyncCredentials } from '../../connection/PowerSyncCredentials.js';
 import { WebsocketClientTransport } from './WebsocketClientTransport.js';
 import { StreamingSyncRequest } from './streaming-sync-types.js';
-;
+
 
 export type BSONImplementation = typeof BSON;
 
@@ -570,17 +570,21 @@ export abstract class AbstractRemote {
       reader.releaseLock();
     };
 
+
+    const stream = new DataStream<T, string>({
+      logger: this.logger,
+      mapLine: mapLine
+    });
+
     abortSignal?.addEventListener('abort', () => {
       closeReader();
+      stream.close();
     });
 
     const decoder = this.createTextDecoder();
     let buffer = '';
 
-    const stream = new DataStream<T, string>({
-      logger: this.logger,
-      mapLine: mapLine
-    });
+
 
     const l = stream.registerListener({
       lowWater: async () => {

packages/node/tests/sync.test.ts

@@ -484,6 +484,7 @@ function defineSyncTests(impl: SyncClientImplementation) {
 
       // Re-open database
       await database.close();
+
       await vi.waitFor(() => expect(syncService.connectedListeners).toHaveLength(0));
       database = await syncService.createDatabase();
       database.connect(new TestConnector(), options);

packages/web/tests/watch.test.ts

@@ -669,32 +669,26 @@ describe('Watch Tests', { sequential: true }, () => {
 
     let notificationCount = 0;
     const dispose = watch.registerListener({
-      onStateChange: (state) => {
+      onStateChange: () => {
         notificationCount++;
       }
     });
     onTestFinished(dispose);
 
     // Wait for the initial load to complete
-    await vi.waitFor(
-      () => {
-        expect(notificationCount).equals(1);
-      },
-      { timeout: 1000 }
-    );
+    await vi.waitFor(() => {
+      expect(notificationCount).equals(1);
+    });
 
     notificationCount = 0; // We want to count the number of state changes after the initial load
 
     // Should only a state change trigger for this operation
     await powersync.execute('INSERT INTO assets(id, make, customer_id) VALUES (uuid(), ?, ?)', ['test', uuid()]);
 
     // We should get an update for the change above
-    await vi.waitFor(
-      () => {
-        expect(notificationCount).equals(1);
-      },
-      { timeout: 1000 }
-    );
+    await vi.waitFor(() => {
+      expect(notificationCount).equals(1);
+    });
 
     // Should not trigger any state change for these operations
     await powersync.execute('INSERT INTO assets(id, make, customer_id) VALUES (uuid(), ?, ?)', ['make1', uuid()]);