fix: stop consuming a mailbox from costing a round trip per message (#159)

guide/sdk.md

@@ -80,5 +80,7 @@ AGENTCOMM_BACKEND_PLUGINS=agentcomm-backend-redis agentcomm send bob hi --backen
 
 `AGENTCOMM_BACKEND_PLUGINS` is a comma/whitespace-separated list of module
 specifiers the CLI imports before resolving `--backend`. Implement
-`Claimable`/`Waitable` too if the store can support atomic claims or push — the Bus
-feature-detects both, no registration needed beyond `Backend` itself.
+`Claimable`/`Waitable`/`Batchable` too if the store can support atomic claims,
+push, or many moves in one operation — the Bus feature-detects all three, no
+registration needed beyond `Backend` itself. `Batchable` is what keeps
+consuming a full mailbox to a single round trip instead of one per message.

src/backends/git.ts

@@ -3,7 +3,7 @@ import { createHash, randomUUID } from 'node:crypto';
 import { promises as fs } from 'node:fs';
 import * as os from 'node:os';
 import * as path from 'node:path';
-import { type Backend, type Claimable, type Message, type Snapshottable } from '../types.js';
+import { type Backend, type Batchable, type Claimable, type Message, type Snapshottable } from '../types.js';
 
 /**
  * GitBackend — the generic "commits are the storage" transport, host-agnostic
@@ -34,7 +34,15 @@ import { type Backend, type Claimable, type Message, type Snapshottable } from '
  *   - `claim` is implemented (Claimable) via optimistic CAS — race-free
  *     shared work queues with zero infrastructure.
  */
-export class GitBackend implements Backend, Claimable, Snapshottable {
+/**
+ * Cap on a single git invocation. A remote that accepts the connection and
+ * then goes quiet leaves `git` blocked forever, which reaches the caller as a
+ * command that simply never returns — indistinguishable from a deadlock
+ * (issue #159). A clear, attributable error is always better.
+ */
+const GIT_TIMEOUT_MS = Math.max(1000, Number(process.env.AGENTCOMM_GIT_TIMEOUT_MS ?? 120_000));
+
+export class GitBackend implements Backend, Batchable, Claimable, Snapshottable {
   /** Each poll is a real fetch — cheap against local remotes, a round trip against hosts. */
   readonly pollIntervalMs = 2000;
 
@@ -98,13 +106,28 @@ export class GitBackend implements Backend, Claimable, Snapshottable {
       });
       const out: Buffer[] = [];
       const err: Buffer[] = [];
+      let timedOut = false;
+      const timer = setTimeout(() => {
+        timedOut = true;
+        child.kill('SIGKILL');
+      }, GIT_TIMEOUT_MS);
+      timer.unref?.();
       child.stdout.on('data', (d: Buffer) => out.push(d));
       child.stderr.on('data', (d: Buffer) => err.push(d));
-      child.on('error', (e) =>
-        reject(e.message.includes('ENOENT') ? new Error('agentcomm: the git+ backends need the `git` binary on PATH') : e),
-      );
+      child.on('error', (e) => {
+        clearTimeout(timer);
+        reject(e.message.includes('ENOENT') ? new Error('agentcomm: the git+ backends need the `git` binary on PATH') : e);
+      });
       child.on('close', (code) => {
-        if (code === 0) resolve(Buffer.concat(out));
+        clearTimeout(timer);
+        if (timedOut) {
+          reject(
+            new Error(
+              `agentcomm: git ${args[0]} on ${this.remote} timed out after ${GIT_TIMEOUT_MS}ms ` +
+                '(unreachable or very slow remote; raise AGENTCOMM_GIT_TIMEOUT_MS if this is normal for your bus)',
+            ),
+          );
+        } else if (code === 0) resolve(Buffer.concat(out));
         else {
           const e = new Error(
             `agentcomm: git ${args[0]} failed (exit ${code}): ${Buffer.concat(err).toString('utf8').trim().slice(0, 400)}`,
@@ -286,6 +309,37 @@ export class GitBackend implements Backend, Claimable, Snapshottable {
     }
   }
 
+  /**
+   * Archive a whole mailbox in ONE commit (issue #159). Consuming N messages
+   * key-by-key is N fetch→commit→push round trips — seconds each, and the
+   * command times out long before the last one lands. Batched, it is one.
+   */
+  async moveMany(moves: { src: string; dst: string }[]): Promise<void> {
+    if (moves.length === 0) return;
+    if (moves.length === 1) return this.move(moves[0]!.src, moves[0]!.dst);
+    for (let attempt = 1; attempt <= 6; attempt++) {
+      const tip = await this.tip();
+      if (tip === null) throw notFound(moves[0]!.src);
+      const add: { key: string; blob: string }[] = [];
+      const remove: string[] = [];
+      for (const { src, dst } of moves) {
+        let blob: string;
+        try {
+          blob = (await this.git(['rev-parse', `${tip}:${this.k(src)}`])).toString('utf8').trim();
+        } catch {
+          continue; // already moved by someone else — not this batch's problem
+        }
+        add.push({ key: dst, blob });
+        remove.push(src);
+      }
+      if (remove.length === 0) return;
+      const message = `agentcomm: archive ${remove.length} message(s) [${randomUUID().slice(0, 8)}]`;
+      if (await this.commitAndPush(tip, message, { add, remove })) return;
+      await sleep(30 * attempt + Math.floor(Math.random() * 80));
+    }
+    throw new Error(`agentcomm: git moveMany kept losing push races — extremely contended bus?`);
+  }
+
   async move(src: string, dst: string): Promise<void> {
     // One commit adds dst and removes src — push lands it atomically.
     for (let attempt = 1; attempt <= 6; attempt++) {

src/backends/socket.ts

@@ -179,7 +179,12 @@ export class SocketBackend implements Backend {
   }
 
   async move(src: string, dst: string): Promise<void> {
-    ok(await this.rpc.call('move', { src, dst }));
+    ok(await this.rpc.call('move', { src, dst, sync: this.syncWrites }));
+  }
+
+  /** Archiving a mailbox is ONE call (issue #159), not one per message. */
+  async moveMany(moves: { src: string; dst: string }[]): Promise<void> {
+    ok(await this.rpc.call('moveMany', { moves, sync: this.syncWrites }));
   }
 
   async info(): Promise<Record<string, unknown>> {

src/bus.ts

@@ -1,5 +1,5 @@
 import { randomUUID } from 'node:crypto';
-import { isClaimable, isWaitable, type Backend, type Message } from './types.js';
+import { isBatchable, isClaimable, isWaitable, type Backend, type Message } from './types.js';
 
 /** How long an explicit status stays sticky before a newer task can refresh it. */
 const EXPLICIT_STICKY_MS = Number(process.env.AGENTCOMM_EXPLICIT_STICKY_MS ?? 15 * 60_000);
@@ -181,6 +181,19 @@ export class Bus {
    * a store that went away mid-run; they stay pending and re-deliver.
    */
   async archive(keys: string[]): Promise<string[]> {
+    if (keys.length === 0) return [];
+    // One store operation for the whole mailbox where the backend can do it
+    // (issue #159): key-by-key, archiving a full inbox is a network round
+    // trip per message and the command times out before it finishes.
+    if (isBatchable(this.backend)) {
+      try {
+        await this.backend.moveMany(keys.map((key) => ({ src: key, dst: readKeyFromInboxKey(key) })));
+        return [];
+      } catch {
+        // fall through: retry key-by-key, so a batch that failed as a whole
+        // still archives whatever it individually can
+      }
+    }
     const failed: string[] = [];
     for (const key of keys) {
       try {