changefeed implementation: fix a memory leak

WOW, that was surprisingly easy.  There was a massive memory leak caused
entirely by this "tracker?.once 'error', (err) ->" event listener not
being removed!

Author: williamstein

src/packages/database/nats/leak-search.ts

@@ -1,30 +1,43 @@
 /*
-Code for testing for memory leaks.
-
-  - just doing queries doesn't leak
-  - as of this writing, creating the changefeed does leak.
-
+Code for testing for memory leaks.  As of this commit, nothing tested for here
+leaks memory in my dev setup.
 
 USAGE:
 
 Run with an account_id from your dev server and pass the expose-gc flag so the
 gc command is defined:
 
 
-ACCOUNT_ID="6aae57c6-08f1-4bb5-848b-3ceb53e61ede"  node  --expose-gc
+ACCOUNT_ID="6aae57c6-08f1-4bb5-848b-3ceb53e61ede" DEBUG=cocalc:* DEBUG_CONSOLE=yes node  --expose-gc
 
 Then do this
 
    a = require('@cocalc/database/nats/leak-search')
    await a.testQueryOnly(50)
    await a.testChangefeed(50)
 
+Do a test multiple times to see if there is a real leak, e.g., the following is GOOD:
+
+  > a.testChangefeed(50)
+  Promise {
+    <pending>,
+    [Symbol(async_id_symbol)]: 47,
+    [Symbol(trigger_async_id_symbol)]: 6
+  }
+  > leaked 5.209536 MB
 
+  > a.testChangefeed(50)
+  Promise {
+    <pending>,
+    [Symbol(async_id_symbol)]: 3167,
+    [Symbol(trigger_async_id_symbol)]: 6
+  }
+  > leaked -0.029184 MB  <--- GOOD!
 */
 
 import { db } from "@cocalc/database";
 import { uuid } from "@cocalc/util/misc";
-//import { delay } from "awaiting";
+import { delay } from "awaiting";
 import { callback2 } from "@cocalc/util/async-utils";
 
 // set env variable to an account_id on your dev server with lots of projects.
@@ -52,12 +65,16 @@ export function cancel(id) {
 }
 
 let pre: any = { heapUsed: 0 };
-function before() {
+async function before() {
+  gc?.();
+  await delay(500);
   gc?.();
   pre = process.memoryUsage();
 }
 
-function after() {
+async function after() {
+  gc?.();
+  await delay(500);
   gc?.();
   const post = process.memoryUsage();
   const leak = (post.heapUsed - pre.heapUsed) / 10 ** 6;
@@ -67,18 +84,18 @@ function after() {
 
 // This leaks horribly
 export async function testChangefeed(n) {
-  before();
+  await before();
   for (let i = 0; i < n; i++) {
     const id = uuid();
     await callback2(create, { id });
     cancel(id);
   }
-  return after();
+  return await after();
 }
 
 // query only does NOT leak
 export async function testQueryOnly(n) {
-  before();
+  await before();
   for (let i = 0; i < n; i++) {
     const d = db();
     await callback2(d.user_query, {
@@ -90,5 +107,5 @@ export async function testQueryOnly(n) {
       account_id: "6aae57c6-08f1-4bb5-848b-3ceb53e61ede",
     });
   }
-  return after();
+  return await after();
 }

src/packages/database/postgres-user-queries.coffee

@@ -1364,6 +1364,11 @@ exports.extend_PostgreSQL = (ext) -> class PostgreSQL extends ext
                 if pg_changefeed == 'projects'
                     tracker_add = (project_id) => feed.insert({project_id:project_id})
                     tracker_remove = (project_id) => feed.delete({project_id:project_id})
+
+                    # Any tracker error means this changefeed is now broken and
+                    # has to be recreated.
+                    tracker_error = () => changes.cb("tracker error - ${err}")
+
                     pg_changefeed =  (db, account_id) =>
                         where  : (obj) =>
                             # Check that this is a project we have read access to
@@ -1381,11 +1386,14 @@ exports.extend_PostgreSQL = (ext) -> class PostgreSQL extends ext
                         init_tracker : (tracker) =>
                             tracker.on "add_user_to_project-#{account_id}", tracker_add
                             tracker.on "remove_user_from_project-#{account_id}", tracker_remove
+                            tracker.once 'error', tracker_error
+
 
                         free_tracker : (tracker) =>
                             dbg("freeing project tracker events")
                             tracker.removeListener("add_user_to_project-#{account_id}", tracker_add)
                             tracker.removeListener("remove_user_from_project-#{account_id}", tracker_remove)
+                            tracker.removeListener("error", tracker_error)
 
 
                 else if pg_changefeed == 'news'
@@ -1423,6 +1431,7 @@ exports.extend_PostgreSQL = (ext) -> class PostgreSQL extends ext
                         return
                     tracker_add = (collab_id) => feed.insert({account_id:collab_id})
                     tracker_remove = (collab_id) => feed.delete({account_id:collab_id})
+                    tracker_error = () => changes.cb("tracker error - ${err}")
                     pg_changefeed = (db, account_id) ->
                         shared_tracker = undefined
                         where : (obj) ->  # test of "is a collab with me"
@@ -1431,10 +1440,12 @@ exports.extend_PostgreSQL = (ext) -> class PostgreSQL extends ext
                             shared_tracker = tracker
                             tracker.on "add_collaborator-#{account_id}", tracker_add
                             tracker.on "remove_collaborator-#{account_id}", tracker_remove
+                            tracker.once 'error', tracker_error
                         free_tracker : (tracker) =>
                             dbg("freeing collab tracker events")
                             tracker.removeListener("add_collaborator-#{account_id}", tracker_add)
                             tracker.removeListener("remove_collaborator-#{account_id}", tracker_remove)
+                            tracker.removeListener("error", tracker_error)
 
 
                 x = pg_changefeed(@, account_id)
@@ -1470,11 +1481,10 @@ exports.extend_PostgreSQL = (ext) -> class PostgreSQL extends ext
                     select : select
                     where  : where
                     watch  : watch
-                    cb     : (err, _feed) =>
+                    cb     : (err, feed) =>
                         if err
                             cb(err)
                             return
-                        feed = _feed
                         feed.on 'change', (x) ->
                             process(x)
                             changes.cb(undefined, x)
@@ -1484,16 +1494,13 @@ exports.extend_PostgreSQL = (ext) -> class PostgreSQL extends ext
                             if tracker? and free_tracker?
                                 dbg("free_tracker")
                                 free_tracker(tracker)
-                            dbg("do NOT free_tracker")
+                            else
+                                dbg("do NOT free_tracker")
                         feed.on 'error', (err) ->
                             changes.cb("feed error - #{err}")
                         @_changefeeds ?= {}
                         @_changefeeds[changes.id] = feed
                         init_tracker?(tracker)
-                        # Any tracker error means this changefeed is now broken and
-                        # has to be recreated.
-                        tracker?.once 'error', (err) ->
-                            changes.cb("tracker error - #{err}")
                         cb()
         ], cb)