core stream/persistent -- be a little more careful with closed state and make warning clearer

- this is an expected warning to happen sometimes
- these code changes probably don't actually fix or break anything; just
  fitting our general style better.

Author: williamstein

src/packages/conat/persist/client.ts

@@ -65,7 +65,7 @@ class PersistStreamClient extends EventEmitter {
       this.close();
       return;
     }
-    if (this.state == "closed") {
+    if (this.isClosed()) {
       return;
     }
     this.socket?.close();
@@ -173,6 +173,8 @@ class PersistStreamClient extends EventEmitter {
     this.emit("changefeed", updates);
   };
 
+  private isClosed = () => this.state == "closed";
+
   close = () => {
     logger.debug("close", this.storage);
     // paths.delete(this.storage.path);
@@ -341,6 +343,10 @@ class PersistStreamClient extends EventEmitter {
     timeout,
     maxWait,
   }: GetAllOpts = {}): AsyncGenerator<StoredMessage[], void, unknown> {
+    if (this.isClosed()) {
+      // done
+      return;
+    }
     const sub = await this.socket.requestMany(null, {
       headers: {
         cmd: "getAll",
@@ -351,6 +357,10 @@ class PersistStreamClient extends EventEmitter {
       timeout,
       maxWait,
     });
+    if (this.isClosed()) {
+      // done with this
+      return;
+    }
     let seq = 0; // next expected seq number for the sub (not the data)
     for await (const { data, headers } of sub) {
       if (headers?.error) {
@@ -384,9 +394,15 @@ class PersistStreamClient extends EventEmitter {
     // This throws with code=503 if something goes wrong due to sequence numbers.
     let messages: StoredMessage[] = [];
     const sub = await this.getAllIter(opts);
+    if (this.isClosed()) {
+      throw Error("closed");
+    }
     for await (const value of sub) {
       messages = messages.concat(value);
     }
+    if (this.isClosed()) {
+      throw Error("closed");
+    }
     return messages;
   };
 

src/packages/conat/sync/core-stream.ts

@@ -261,6 +261,10 @@ export class CoreStream<T = any> extends EventEmitter {
     return await this.persistClient.config({ config });
   };
 
+  private isClosed = () => {
+    return this.client === undefined;
+  };
+
   close = () => {
     logger.debug("close", this.name);
     delete this.client;
@@ -296,23 +300,20 @@ export class CoreStream<T = any> extends EventEmitter {
     }
     await until(
       async () => {
-        if (this.client == null) {
+        if (this.isClosed()) {
           return true;
         }
         let messages: StoredMessage[] = [];
         let changes: (SetOperation | DeleteOperation | StoredMessage)[] = [];
         let changefeed: any = undefined;
         try {
           changefeed = await this.persistClient.changefeed();
+          if (this.isClosed()) {
+            return true;
+          }
           (async () => {
             try {
               for await (const updates of changefeed) {
-                //                 console.log(
-                //                   "getAllFromPersist-changefeed-update",
-                //                   this.client.id,
-                //                   this.storage.path,
-                //                   JSON.stringify(updates.map(({ seq }) => seq)),
-                //                 );
                 changes = changes.concat(updates);
               }
             } catch {}
@@ -322,9 +323,12 @@ export class CoreStream<T = any> extends EventEmitter {
             start_seq,
           });
         } catch (err) {
+          if (this.isClosed()) {
+            return true;
+          }
           if (!process.env.COCALC_TEST_MODE) {
             console.log(
-              `WARNING: getAllFromPersist - failed -- ${err}, code=${err.code}, service=${this.service}, storage=${JSON.stringify(this.storage)}`,
+              `WARNING: getAllFromPersist - failed -- ${err}, code=${err.code}, service=${this.service}, storage=${JSON.stringify(this.storage)} -- will retry`,
             );
           }
           if (err.code == 503 || err.code == 408) {