changefeeds -- make server more defensive and client less aggressive

Author: williamstein

src/packages/nats/changefeed/server.ts

@@ -20,8 +20,13 @@ export const MIN_LIFETIME = 30 * 1000;
 export const MIN_HEARTBEAT = 5000;
 export const MAX_HEARTBEAT = 120000;
 export const MAX_CHANGEFEEDS_PER_ACCOUNT = parseInt(
-  process.env.COCALC_MAX_CHANGEFEEDS_PER_ACCOUNT ?? "150",
+  process.env.MAX_CHANGEFEEDS_PER_ACCOUNT ?? "100",
 );
+
+export const MAX_CHANGEFEEDS_PER_SERVER = parseInt(
+  process.env.MAX_CHANGEFEEDS_PER_SERVER ?? "5000",
+);
+
 export const LAST_CHUNK = "last-chunk";
 
 const logger = getLogger("changefeed:server");
@@ -52,6 +57,16 @@ let terminated = false;
 let sub: Subscription | null = null;
 export async function init(db) {
   logger.debug("starting changefeed server");
+  logger.debug({
+    DEFAULT_LIFETIME,
+    MAX_LIFETIME,
+    MIN_LIFETIME,
+    MIN_HEARTBEAT,
+    MAX_HEARTBEAT,
+    MAX_CHANGEFEEDS_PER_ACCOUNT,
+    MAX_CHANGEFEEDS_PER_SERVER,
+    SUBJECT,
+  });
   changefeedMainLoop(db);
   renewMainLoop();
 }
@@ -230,6 +245,18 @@ async function handleMessage(mesg, db) {
     );
     return;
   }
+  if (numChangefeeds >= MAX_CHANGEFEEDS_PER_SERVER) {
+    logger.debug("numChangefeeds >= MAX_CHANGEFEEDS_PER_SERVER", {
+      numChangefeeds,
+      MAX_CHANGEFEEDS_PER_SERVER,
+    });
+    // this will just cause the client to make another attempt, hopefully
+    // to another server
+    respond(
+      `There is a limit of ${MAX_CHANGEFEEDS_PER_SERVER} changefeeds per server`,
+    );
+    return;
+  }
 
   let { heartbeat } = request;
   const lifetime = getLifetime(request);

src/packages/sync/table/synctable.ts

@@ -721,7 +721,7 @@ export class SyncTable extends EventEmitter {
   }
 
   private create_changefeed_connection = async (): Promise<any[]> => {
-    let delay_ms: number = 500;
+    let delay_ms: number = 3000;
     while (true) {
       this.close_changefeed();
       if (
@@ -757,13 +757,11 @@ export class SyncTable extends EventEmitter {
         }
         // This can happen because we might suddenly NOT be ready
         // to query db immediately after we are ready...
-        console.warn(
-          `WARNING: ${this.table} -- failed to create changefeed connection -- ${err}; will retry`,
+        console.log(
+          `WARNING: ${this.table} -- failed to create changefeed connection; will retry in ${delay_ms}ms -- ${err}`,
         );
         await delay(delay_ms);
-        if (delay_ms < 7000) {
-          delay_ms *= 1.3;
-        }
+        delay_ms = Math.min(20000, delay_ms * 1.25);
       }
     }
   };