watch debouncing

Author: jbolda

package-lock.json

@@ -40,7 +40,7 @@
     "effection": "^4.0.0",
     "@effectionx/context-api": "^0.2.1",
     "@effectionx/process": "^0.6.2",
-    "@effectionx/stream-helpers": "^0.4.1",
+    "@effectionx/stream-helpers": "^0.5.1",
     "@effectionx/timebox": "^0.3.1",
     "chokidar": "^5.0.0",
     "picomatch": "^4.0.3"

packages/server/package.json

@@ -35,6 +35,7 @@ export type ServiceGraph<
     [service in keyof S]: ServiceDefinition<keyof S, T>;
   };
   serviceUpdates?: Stream<ServiceUpdate, unknown> | undefined;
+  serviceChanges?: Stream<ServiceUpdate, unknown> | undefined;
   // map of service name => listening port (when the service exposes one)
   servicePorts?: Map<string, number> | undefined;
 };
@@ -122,7 +123,12 @@ export function useServiceGraph<
         );
       }
 
-      const watcher = yield* useWatcher(effectiveServices);
+      const watcher = yield* useWatcher(
+        effectiveServices,
+        options?.watchDebounce
+          ? { watchDebounce: options.watchDebounce }
+          : undefined
+      );
 
       const status = new Map<
         string,
@@ -156,7 +162,7 @@ export function useServiceGraph<
 
       yield* spawn(function* () {
         // restart propagation to dependents is handled by useWatcher
-        for (let restartService of yield* each(watcher.serviceUpdates)) {
+        for (let restartService of yield* each(watcher.serviceChanges)) {
           bumpService(restartService.service);
           yield* each.next();
         }
@@ -219,6 +225,7 @@ export function useServiceGraph<
         yield* provide({
           services: services as S,
           serviceUpdates: watcher?.serviceUpdates,
+          serviceChanges: watcher?.serviceChanges,
           servicePorts,
         });
       } finally {

packages/server/src/services.ts

@@ -5,44 +5,25 @@ import {
   createSignal,
   each,
   type Operation,
-  race,
   resource,
-  sleep,
   spawn,
   type Stream,
   until,
 } from "effection";
 import picomatch, { type Matcher } from "picomatch";
-
-export function debounce<T, R>(
-  ms: number
-): (stream: Stream<T, R>) => Stream<T, R> {
-  return (stream) => ({
-    *[Symbol.iterator]() {
-      let subscription = yield* stream;
-      return {
-        *next() {
-          let next = yield* subscription.next();
-          while (true) {
-            let result = yield* race([sleep(ms), subscription.next()]);
-            if (!result) {
-              return next;
-            } else {
-              next = result;
-            }
-          }
-        },
-      };
-    },
-  });
-}
+import { filter } from "@effectionx/stream-helpers";
 
 export type ServiceUpdate = { service: string; path: string };
 
 export function useWatcher(
-  services?: Record<string, { dependsOn?: { restart?: readonly string[] } }>
+  services?: Record<
+    string,
+    { dependsOn?: { restart?: readonly string[] }; watchDebounce?: number }
+  >,
+  options?: { watchDebounce?: number }
 ): Operation<{
   serviceUpdates: Stream<{ service: string; path: string }, void>;
+  serviceChanges: Stream<{ service: string; path: string }, void>;
   add: (service: string, paths: string[]) => void;
 }> {
   return resource(function* (provide) {
@@ -119,8 +100,30 @@ export function useWatcher(
       }
     });
 
+    const debounceMs =
+      options?.watchDebounce !== undefined ? options.watchDebounce : 250;
+    const serviceTimers = {} as Record<string, number>;
+    const debouncedServiceChanges = filter<ServiceUpdate>(function* (
+      updateStream
+    ) {
+      const now = performance.now();
+      if (
+        serviceTimers[updateStream.service] &&
+        now - serviceTimers[updateStream.service] < debounceMs
+      ) {
+        return false;
+      } else {
+        serviceTimers[updateStream.service] = now;
+        return true;
+      }
+    });
+
     try {
-      yield* provide({ serviceUpdates, add });
+      yield* provide({
+        serviceUpdates,
+        add,
+        serviceChanges: debouncedServiceChanges(serviceUpdates),
+      });
     } finally {
       yield* until(watcher.close());
     }

packages/server/src/watch.ts

@@ -1,6 +1,6 @@
 import { it } from "node:test";
 import assert from "node:assert";
-import { run, suspend, sleep, until, spawn, resource } from "effection";
+import { run, suspend, sleep, until, spawn, resource, ensure } from "effection";
 import * as fs from "node:fs/promises";
 import path from "node:path";
 import os from "node:os";
@@ -41,12 +41,12 @@ it("restarts services on watched file change and restarts dependents", async ()
 
       try {
         const services = yield* op();
-        // subscribe to the immediate serviceUpdates stream and wait for the first update
-        if (!services.serviceUpdates)
-          throw new Error("serviceUpdates not available");
-        const subscription = yield* services.serviceUpdates;
+        // subscribe to the immediate raw serviceChanges stream and wait for the first update
+        if (!services.serviceChanges)
+          throw new Error("serviceChanges not available");
+        const subscription = yield* services.serviceChanges;
 
-        // wait for the first update (will occur after the test touches the file)
+        // wait for the first raw update (will occur after the test touches the file)
         const first = yield* subscription.next();
         updates.push(String((first.value as { service: string }).service));
       } catch (e) {
@@ -210,3 +210,84 @@ it("restarts transitive dependents when watched service changes", async () => {
   assert(startCounts.b >= 2, "b should have been restarted as dependent");
   assert(startCounts.c >= 2, "c should have been restarted as dependent of b");
 });
+
+it("debounces rapid changes per service", async () => {
+  const prefix = path.join(os.tmpdir(), "sim-watch-debounce-");
+  const dir = await fs.mkdtemp(prefix);
+  const trigger = path.join(dir, "trigger.txt");
+  await fs.writeFile(trigger, "initial");
+
+  const updates: string[] = [];
+  let rawCount = 0;
+
+  await run(function* () {
+    yield* spawn(function* () {
+      const op = useServiceGraph(
+        {
+          a: {
+            watch: [dir],
+            operation: resource<void>(function* (provide) {
+              yield* provide();
+            }),
+          },
+        },
+        { watch: true, watchDebounce: 150 }
+      );
+
+      try {
+        const services = yield* op();
+        if (!services.serviceUpdates || !services.serviceChanges)
+          throw new Error("service streams not available");
+        const debSub = yield* services.serviceUpdates;
+        const rawSub = yield* services.serviceChanges;
+
+        // collect debounced updates
+        yield* spawn(function* () {
+          while (true) {
+            const n = yield* debSub.next();
+            if (n.done) break;
+            updates.push((n.value as { service: string }).service);
+          }
+        });
+
+        // count raw updates (should reflect every write)
+        yield* spawn(function* () {
+          while (true) {
+            const n = yield* rawSub.next();
+            if (n.done) break;
+            if ((n.value as { service: string }).service === "a") rawCount++;
+          }
+        });
+      } catch (e) {
+        throw e;
+      }
+
+      yield* suspend();
+    });
+
+    // ensure watcher attached
+    yield* sleep(0);
+
+    // write multiple times rapidly
+    yield* until(fs.writeFile(trigger, "changed-1"));
+    yield* sleep(10);
+    yield* until(fs.writeFile(trigger, "changed-2"));
+    yield* sleep(10);
+    yield* until(fs.writeFile(trigger, "changed-3"));
+
+    yield* ensure(() => until(fs.rm(dir, { recursive: true, force: true })));
+    // wait longer than debounce window
+    yield* sleep(300);
+  });
+
+  // we expect the rapid writes to coalesce: there should be at least one
+  // raw and at least one debounced update, and debounced updates should be
+  // fewer than the number of writes (3)
+  assert(rawCount >= 1, `expected at least 1 raw update, got ${rawCount}`);
+  assert(updates.length >= 1, "expected at least one debounced update");
+  const aCount = updates.filter((u) => u === "a").length;
+  assert(
+    aCount < 3,
+    `expected debounced updates to be fewer than writes (3), got ${aCount}`
+  );
+});