fix(herdr-core): keep an Ovis unpin whose label write failed after its terminal came back a shell

changelog.d/herdr-core-failed-unpin-shell.md

@@ -0,0 +1,3 @@
+### Fixed
+- herdr-core: an agent unpinned in Ovis while its label could not be written to the Herdr app no
+  longer comes back pinned after its terminal returned as a shell and an agent runs there again.

server/herdr-core/pin-label.test.ts

@@ -1,5 +1,5 @@
 import { afterEach, describe, expect, it } from "bun:test";
-import { mkdtempSync, rmSync } from "node:fs";
+import { mkdirSync, mkdtempSync, readFileSync, rmSync } from "node:fs";
 import { tmpdir } from "node:os";
 import { join } from "node:path";
 import { MAX_PINS } from "../../shared/preferences.ts";
@@ -337,6 +337,215 @@ describe("bridgePins", () => {
     expect(core.written).toEqual([{ terminal: "t1", labels: { "herdr.pinned": "1" } }, { terminal: "t2", labels: { "herdr.pinned": "1" } }]);
   });
 
+  it("keeps an unpin made in Ovis whose label write failed, of a terminal that came back a shell, once an agent runs there again", async () => {
+    const dir = stateDir();
+    const preferences = store(dir, ["local:t1"]);
+    const terminals: Held[] = [{ address: "t1", state: "live", labels: { "herdr.pinned": "1" } }];
+    const core = fakeCore(terminals);
+    const pins = bridge(core, preferences, dir);
+    await core.opened();
+    await pins.idle();
+    // hc is away: the unpin's label write fails, and t1's agent quits meanwhile
+    core.state.failing = true;
+    preferences.set({ settings: {}, pins: [] });
+    await pins.idle();
+    terminals[0]!.status = null;
+    await core.drop();
+    await pins.idle();
+    core.state.failing = false;
+    core.tell({ type: "status", terminal: "t1", data: { status: AGENT } });
+    await pins.idle();
+    expect(preferences.get().pins).toEqual([]);
+    expect(terminals[0]!.labels).toEqual({});
+    expect(core.written).toEqual([{ terminal: "t1", labels: { "herdr.pinned": "" } }]);
+  });
+
+  it("keeps a pin the Herdr app made again after an Ovis unpin whose label write failed, while the terminal was a shell", async () => {
+    const dir = stateDir();
+    const preferences = store(dir, ["local:t1"]);
+    const terminals: Held[] = [{ address: "t1", state: "live", labels: { "herdr.pinned": "1" } }];
+    const core = fakeCore(terminals);
+    const pins = bridge(core, preferences, dir);
+    await core.opened();
+    await pins.idle();
+    core.state.failing = true;
+    preferences.set({ settings: {}, pins: [] });
+    await pins.idle();
+    // t1's agent quits; the Herdr app unpins it, then pins it again
+    core.tell({ type: "status", terminal: "t1", data: { status: null } });
+    core.tell({ type: "labels", terminal: "t1", data: { labels: {} } });
+    core.tell({ type: "labels", terminal: "t1", data: { labels: { "herdr.pinned": "1" } } });
+    core.state.failing = false;
+    core.tell({ type: "status", terminal: "t1", data: { status: AGENT } });
+    await pins.idle();
+    expect(preferences.get().pins).toEqual(["local:t1"]);
+    expect(core.written).toEqual([]);
+  });
+
+  it("takes a pin whose write landed but whose answer failed for no pending change of Ovis's", async () => {
+    const dir = stateDir();
+    const preferences = store(dir, []);
+    const core = fakeCore([{ address: "t1", state: "live", labels: {} }, { address: "t2", state: "live", labels: {} }]);
+    const pins = bridge(core, preferences, dir);
+    await core.opened();
+    await pins.idle();
+    const answer = Promise.withResolvers<void>();
+    core.state.answer = answer.promise;
+    const reached = core.writing();
+    preferences.set({ settings: {}, pins: ["local:t1"] });
+    await reached;
+    core.state.answer = null;
+    answer.reject(new Error("reply lost"));
+    await pins.idle();
+    // t1's agent quits; the Agents list drops the shell's pin as t2 is pinned
+    core.tell({ type: "status", terminal: "t1", data: { status: null } });
+    preferences.set({ settings: {}, pins: ["local:t2"] });
+    await pins.idle();
+    core.tell({ type: "status", terminal: "t1", data: { status: AGENT } });
+    await pins.idle();
+    expect(preferences.get().pins).toEqual(["local:t2", "local:t1"]);
+  });
+
+  it("takes an Ovis unpin that failed after Ovis pinned again for no pending change", async () => {
+    const dir = stateDir();
+    const preferences = store(dir, ["local:t1"]);
+    const core = fakeCore([{ address: "t1", state: "live", labels: { "herdr.pinned": "1" } }, { address: "t2", state: "live", labels: {} }]);
+    const gate = Promise.withResolvers<void>();
+    const hc = core.hc as unknown as { call: (method: string, params: unknown) => Promise<unknown> };
+    const call = hc.call;
+    const hanging = { on: false };
+    const reached = Promise.withResolvers<void>();
+    hc.call = async (method, params) => {
+      if (!hanging.on) return call(method, params);
+      reached.resolve();
+      await gate.promise;
+      throw new Error("hc is not answering");
+    };
+    const pins = bridge(core, preferences, dir);
+    await core.opened();
+    await pins.idle();
+    // Ovis unpins t1 and its write hangs; Ovis pins it again, then t1's agent quits and the write fails
+    hanging.on = true;
+    preferences.set({ settings: {}, pins: [] });
+    await reached.promise;
+    preferences.set({ settings: {}, pins: ["local:t1"] });
+    core.tell({ type: "status", terminal: "t1", data: { status: null } });
+    gate.resolve();
+    await pins.idle();
+    hanging.on = false;
+    // the Agents list drops the shell's pin as t2 is pinned
+    preferences.set({ settings: {}, pins: ["local:t2"] });
+    await pins.idle();
+    core.tell({ type: "status", terminal: "t1", data: { status: AGENT } });
+    await pins.idle();
+    expect(preferences.get().pins).toEqual(["local:t2", "local:t1"]);
+  });
+
+  it("keeps a pin made again in Ovis while the retry of its failed unpin waited, once the retry fails too", async () => {
+    const dir = stateDir();
+    const preferences = store(dir, ["local:t1"]);
+    const core = fakeCore([{ address: "t1", state: "live", labels: { "herdr.pinned": "1" } }, { address: "t2", state: "live", labels: {} }]);
+    const gate = Promise.withResolvers<void>();
+    const reached = Promise.withResolvers<void>();
+    const hc = core.hc as unknown as { call: (method: string, params: unknown) => Promise<unknown> };
+    const call = hc.call;
+    const mode = { now: "pass" as "pass" | "fail" | "hang" };
+    hc.call = async (method, params) => {
+      if (mode.now === "pass") return call(method, params);
+      if (mode.now === "hang") { reached.resolve(); await gate.promise; }
+      else mode.now = "hang";
+      throw new Error("hc is not answering");
+    };
+    const pins = bridge(core, preferences, dir);
+    await core.opened();
+    await pins.idle();
+    // the unpin fails; its retry hangs while Ovis pins t1 again and t1's agent quits, then fails too
+    mode.now = "fail";
+    preferences.set({ settings: {}, pins: [] });
+    await reached.promise;
+    preferences.set({ settings: {}, pins: ["local:t1"] });
+    core.tell({ type: "status", terminal: "t1", data: { status: null } });
+    gate.resolve();
+    await pins.idle();
+    mode.now = "pass";
+    // the Agents list drops the shell's pin as t2 is pinned
+    preferences.set({ settings: {}, pins: ["local:t2"] });
+    await pins.idle();
+    core.tell({ type: "status", terminal: "t1", data: { status: AGENT } });
+    await pins.idle();
+    expect(preferences.get().pins).toEqual(["local:t2", "local:t1"]);
+    expect(core.written).toEqual([{ terminal: "t2", labels: { "herdr.pinned": "1" } }]);
+  });
+
+  for (const first of ["hangs", "fails and its retry hangs"] as const) {
+    it(`keeps a pin made again in Ovis when an unpin's write ${first}, and the shell's row is pruned before it fails`, async () => {
+      const dir = stateDir();
+      const preferences = store(dir, ["local:t1"]);
+      const core = fakeCore([{ address: "t1", state: "live", labels: { "herdr.pinned": "1" } }, { address: "t2", state: "live", labels: {} }]);
+      const gate = Promise.withResolvers<void>();
+      const reached = Promise.withResolvers<void>();
+      const hc = core.hc as unknown as { call: (method: string, params: unknown) => Promise<unknown> };
+      const call = hc.call;
+      const mode = { now: "pass" as "pass" | "fail" | "hang" };
+      hc.call = async (method, params) => {
+        if (mode.now === "pass") return call(method, params);
+        if (mode.now === "hang") { reached.resolve(); await gate.promise; }
+        else mode.now = "hang";
+        throw new Error("hc is not answering");
+      };
+      const pins = bridge(core, preferences, dir);
+      await core.opened();
+      await pins.idle();
+      mode.now = first === "hangs" ? "hang" : "fail";
+      preferences.set({ settings: {}, pins: [] });
+      await reached.promise;
+      // Ovis pins t1 again; its agent quits and the Agents list drops the shell's pin as t2 is pinned
+      preferences.set({ settings: {}, pins: ["local:t1"] });
+      core.tell({ type: "status", terminal: "t1", data: { status: null } });
+      mode.now = "pass";
+      preferences.set({ settings: {}, pins: ["local:t2"] });
+      mode.now = "hang";
+      gate.resolve();
+      await pins.idle();
+      mode.now = "pass";
+      await pins.idle();
+      core.tell({ type: "status", terminal: "t1", data: { status: AGENT } });
+      await pins.idle();
+      expect(preferences.get().pins).toEqual(["local:t2", "local:t1"]);
+      expect(core.written.filter((write) => write.terminal === "t1")).toEqual([]);
+    });
+  }
+
+  it("saves the pending mark of a failed Ovis unpin on a later failed write when the first save failed", async () => {
+    const dir = stateDir();
+    const preferences = store(dir, ["local:t1"]);
+    const core = fakeCore([{ address: "t1", state: "live", labels: { "herdr.pinned": "1" } }]);
+    const hc = core.hc as unknown as { call: (method: string, params: unknown) => Promise<unknown> };
+    const call = hc.call;
+    let attempts = 0;
+    const second = Promise.withResolvers<void>();
+    hc.call = async (method, params) => {
+      if (!core.state.failing) return call(method, params);
+      attempts += 1;
+      if (attempts === 2) second.resolve();
+      throw new Error("hc is not answering");
+    };
+    const pins = bridge(core, preferences, dir);
+    await core.opened();
+    await pins.idle();
+    // the first failed unpin cannot save its mark: the pins file's temporary path is taken
+    const blocker = join(dir, `herdr-core-pins.json.${process.pid}.tmp`);
+    mkdirSync(blocker);
+    core.state.failing = true;
+    preferences.set({ settings: {}, pins: [] });
+    await pins.idle();
+    rmSync(blocker, { recursive: true });
+    await second.promise;
+    await pins.idle();
+    pins.stop();
+    expect(JSON.parse(readFileSync(join(dir, "herdr-core-pins.json"), "utf8")).pending).toEqual(["c1/t1"]);
+  });
+
   it("keeps the last of a quick pin and unpin, whatever order hc's answer and event come in", async () => {
     const dir = stateDir();
     const preferences = store(dir, []);

server/herdr-core/pin-label.ts

@@ -43,7 +43,7 @@ export function bridgePins(hc: Hc, store: PinStore, machineId: string, stateDir:
    * `<core id>/<terminal>` → the pin both sides last agreed on; `shells`, the agreed terminals seen
    * without an agent since, until an agent runs there again (kept on disk, as a restart may come between)
    */
-  const { agreed, shells } = readAgreed(agreedPath);
+  const { agreed, shells, pending } = readAgreed(agreedPath);
   /** each live terminal of the current core: whether its label says pinned, as last heard */
   const live = new Map<string, boolean>();
   /**
@@ -65,16 +65,17 @@ export function bridgePins(hc: Hc, store: PinStore, machineId: string, stateDir:
 
   const saveAgreed = () => {
     try {
-      writeStateDocument(agreedPath, `${JSON.stringify({ ...Object.fromEntries(agreed), [SHELLS]: [...shells] })}\n`, false);
+      writeStateDocument(agreedPath, `${JSON.stringify({ ...Object.fromEntries(agreed), [SHELLS]: [...shells], [PENDING]: [...pending] })}\n`, false);
       unsaved = false;
     } catch (error) {
       unsaved = true;
       console.warn(`herdr-core: could not save ${agreedPath}:`, error);
     }
   };
   const setAgreed = (address: string, pinned: boolean) => {
+    const settled = pending.delete(agreedKey(address));
     if (agreed.get(agreedKey(address)) === pinned) {
-      if (unsaved) saveAgreed();
+      if (unsaved || settled) saveAgreed();
       return;
     }
     agreed.set(agreedKey(address), pinned);
@@ -147,6 +148,18 @@ export function bridgePins(hc: Hc, store: PinStore, machineId: string, stateDir:
         // a write hc refused sends no event back
         const index = queued.indexOf(echo);
         if (index >= 0) queued.splice(index, 1);
+        // Ovis's change stays Ovis's, even once the terminal is a shell whose row the list prunes:
+        // pending while Ovis's list differs from what was agreed (a newer change may have taken it back),
+        // never for a write that landed
+        const key = agreedKey(address);
+        // a shell's row the list may have pruned tells nothing of Ovis's change
+        const differs = agreed.has(key) && agents.has(address) && listed(address) !== agreed.get(key);
+        // saved again while an earlier save failed, so a restart still finds the mark
+        if (!echo.heard && (differs !== pending.has(key) || unsaved)) {
+          if (differs) pending.add(key);
+          else pending.delete(key);
+          saveAgreed();
+        }
         retry(address);
       }
     });
@@ -156,6 +169,7 @@ export function bridgePins(hc: Hc, store: PinStore, machineId: string, stateDir:
 
   /** A terminal's label heard: settled against what both sides last agreed on. */
   const heard = (address: string, labelPinned: boolean) => {
+    const before = live.get(address);
     live.set(address, labelPinned);
     // the event of a write of ours, early or late: no change of the app's
     const queued = echoes.get(address);
@@ -171,14 +185,17 @@ export function bridgePins(hc: Hc, store: PinStore, machineId: string, stateDir:
       // a label neither of our writes set: the app's, and what we sent no longer tells its order
       queued.length = 0;
     }
+    // the app changed the pin after Ovis's failed write: its change is the last one
+    if (before !== undefined && before !== labelPinned && pending.delete(agreedKey(address))) saveAgreed();
     // only an agent is in the Agents list: a shell's pin is never pruned from it as an unpin
     if (!agents.has(address)) return;
     refused.delete(address);
     const base = agreed.get(agreedKey(address));
     const ovis = listed(address);
     if (ovis || !labelPinned) leftOut.delete(address);
-    // back from a shell, Ovis's list may have pruned the row: only a pin there tells Ovis's side
-    const shell = shells.has(agreedKey(address)) && labelPinned === base;
+    // back from a shell, Ovis's list may have pruned the row: only a pin there tells Ovis's side,
+    // unless Ovis changed it itself and the label write failed
+    const shell = shells.has(agreedKey(address)) && labelPinned === base && !pending.has(agreedKey(address));
     const wanted = base === undefined || leftOut.has(address) || shell ? labelPinned || ovis : labelPinned !== base ? labelPinned : ovis;
     if (ovis !== wanted && !list(address, wanted)) {
       refused.add(address);
@@ -227,14 +244,22 @@ export function bridgePins(hc: Hc, store: PinStore, machineId: string, stateDir:
     clearTimeout(retries.get(address)?.timer);
     retries.delete(address);
     const shell = shells.delete(agreedKey(address));
-    if (agreed.delete(agreedKey(address)) || shell) saveAgreed();
+    const unwritten = pending.delete(agreedKey(address));
+    if (agreed.delete(agreedKey(address)) || shell || unwritten) saveAgreed();
   };
 
   const unsubscribe = store.subscribe((preferences) => {
     // a change a full or unsaved list left out gets another look, even one that would start the list
     for (const address of [...refused]) heard(address, live.get(address) ?? false);
     // no browser has shared its pins yet: that is no unpin
     if (preferences.pins === undefined) return;
+    // a change in Ovis's list taken back before its label was written leaves nothing pending
+    let undone = false;
+    for (const address of live.keys()) {
+      const key = agreedKey(address);
+      if (agents.has(address) && pending.has(key) && listed(address) === agreed.get(key)) undone = pending.delete(key);
+    }
+    if (undone) saveAgreed();
     for (const address of live.keys()) writeLabel(address);
   });
 
@@ -287,7 +312,7 @@ export function bridgePins(hc: Hc, store: PinStore, machineId: string, stateDir:
       leftOut.clear();
       const shown = new Set(result.terminals.filter((terminal) => terminal.state === "live").map((terminal) => agreedKey(terminal.address)));
       let pruned = false;
-      for (const key of agreed.keys()) if (key.startsWith(`${core}/`) && !shown.has(key)) { agreed.delete(key); shells.delete(key); pruned = true; }
+      for (const key of agreed.keys()) if (key.startsWith(`${core}/`) && !shown.has(key)) { agreed.delete(key); shells.delete(key); pending.delete(key); pruned = true; }
       if (pruned) saveAgreed();
       // every terminal known before any is settled, so a list started for one holds the others' pins
       for (const terminal of result.terminals) {
@@ -321,18 +346,20 @@ export function bridgePins(hc: Hc, store: PinStore, machineId: string, stateDir:
 
 /** the key in herdr-core-pins.json listing `shells`; never a terminal's, which holds a `/` */
 const SHELLS = "shells";
+/** the key listing agreed terminals whose last label write of Ovis's change failed */
+const PENDING = "pending";
 
-function readAgreed(path: string): { agreed: Map<string, boolean>; shells: Set<string> } {
+function readAgreed(path: string): { agreed: Map<string, boolean>; shells: Set<string>; pending: Set<string> } {
   try {
     const read = readStateDocument(path);
     if (read.kind === "json" && isRecord(read.value)) {
       const agreed = new Map(Object.entries(read.value).filter((entry): entry is [string, boolean] => typeof entry[1] === "boolean"));
       const listed = read.value[SHELLS];
-      const shells = Array.isArray(listed) ? listed.filter((key): key is string => typeof key === "string" && agreed.has(key)) : [];
-      return { agreed, shells: new Set(shells) };
+      const keys = (value: unknown) => Array.isArray(value) ? value.filter((key): key is string => typeof key === "string" && agreed.has(key)) : [];
+      return { agreed, shells: new Set(keys(listed)), pending: new Set(keys(read.value[PENDING])) };
     }
   } catch (error) {
     console.warn(`herdr-core: could not read ${path}; pins are settled as if first seen:`, error);
   }
-  return { agreed: new Map(), shells: new Set() };
+  return { agreed: new Map(), shells: new Set(), pending: new Set() };
 }