Updates to Swarm Disk

docs/reference/environment-and-posix-appendix.md

@@ -39,6 +39,7 @@ The list below is one **bullet per variable** in the form **name — component
 - `BARE_OS_BOOT_TIMEOUT_MS` — Booter — Wall-clock budget for peer wait + network boot (default `60000`)
 - `BARE_OS_MBR_READ_TIMEOUT_MS` — Booter (host) — Max wait in milliseconds for replicated **block 0** / MBR before boot fails (**≥ 3000**, hard cap **600000**); when unset, defaults to **60000** unless adaptive mode adjusts it.
 - `BARE_OS_MBR_READ_TIMEOUT_ADAPTIVE` — Booter — When **`1`** / **`true`** and **`BARE_OS_MBR_READ_TIMEOUT_MS`** is unset, adjusts the MBR wait from live **peer count** (sparse swarms get a longer budget; multi-peer sessions may use a shorter one).
+- `BARE_OS_MBR_READ_REBROADCAST_MS` — Booter — While waiting for **block 0** / MBR, re-send read requests to all peers on this interval (**default `5000`**, clamp **`2000`**–**`60000`** **`ms`**). Set **`0`** / **`false`** to disable periodic rebroadcast (initial and per-peer-join sends only).
 - `BARE_OS_DISK_OS_SEARCH_THROTTLE_MS` — Booter / **`disk.os`** — Optional **0–500** ms delay between scanning the system Hyperdrive and each **`auxiliaryDrives`** entry during peer **`searchLocal`** fan-out (default **0**); softens load on large offline mirrors.
 - `BARE_OS_DISK_OS_PATH_MANIFEST` — Booter / **`disk.os`** — Absolute path on the **system** Hyperdrive to a JSON manifest (**default** **`/etc/bare-os/path-manifest.json`**). The file may be **`{ "schema": 1, "paths": ["/bin/sh", …] }`** or a bare JSON array of path strings. Substring matches from the manifest are merged **before** the primary drive’s recursive listing. Set to **empty** to disable manifest acceleration (full listing only on the primary drive).
 - `BARE_OS_DISK_OS_SEARCH_MANIFEST_ONLY` — Booter — When **`1`** / **`true`**, `**searchLocal` skips the primary drive’s recursive `list()` scan** and uses only manifest hits on that drive (auxiliary drives are still listed). For tests and operator-tuned images with a complete manifest.

packages/bare-os-booter/lib/swarm-disk.js

@@ -80,6 +80,23 @@ export function mbrReadTimeoutMsForDisk(disk) {
   return MBR_READ_TIMEOUT_MS_DEFAULT
 }
 
+/**
+ * Interval (ms) to re-send pending block read requests to all peers while waiting for MBR / block data.
+ * **`0`** disables periodic rebroadcast (initial + per-connect sends only).
+ * Override **`BARE_OS_MBR_READ_REBROADCAST_MS`** (defaults **5000**, clamp **2000**–**60000**).
+ * @param {Record<string, string | undefined> | null | undefined} env
+ */
+export function mbrReadRebroadcastMsFromEnv(env) {
+  const raw = String(env?.BARE_OS_MBR_READ_REBROADCAST_MS ?? '').trim()
+  if (raw === '0' || raw.toLowerCase() === 'false') return 0
+  const n = Math.floor(Number(raw))
+  if (Number.isFinite(n) && n >= 0) {
+    if (n === 0) return 0
+    return Math.min(60_000, Math.max(2_000, n))
+  }
+  return 5_000
+}
+
 export class SwarmDisk {
   constructor() {
     this.localRAM = new Map()
@@ -223,9 +240,22 @@ export class SwarmDisk {
           reject(new Error(m.error || 'swarm-disk rpc failed'))
         }
       })
-      /** @type {{ chan: { messages: { send: (m: unknown) => void }[] } }} */ (
+      const chan = /** @type {{ chan: { messages: { send: (m: unknown) => void }[] } }} */ (
         peer
-      ).chan.messages[5].send({ id, module, method, args })
+      ).chan
+      void chan.fullyOpened().then((opened) => {
+        if (!opened) return
+        if (!this.pendingRpc.has(id)) return
+        try {
+          chan.messages[5].send({ id, module, method, args })
+        } catch (err) {
+          this.pendingRpc.delete(id)
+          clearTimeout(to)
+          reject(
+            err instanceof Error ? err : new Error((err && err.message) || String(err))
+          )
+        }
+      })
     })
   }
 
@@ -595,15 +625,7 @@ export class SwarmDisk {
 
     const pendingIndices = [...this.pendingReads.keys()]
     for (const idx of pendingIndices) {
-      void peer.chan.fullyOpened().then(() => {
-        if (!this.peers.has(peer)) return
-        if (!this.pendingReads.has(idx)) return
-        try {
-          peer.chan.messages[0].send(idx)
-        } catch {
-          /* ignore — bad channel; other peers may still satisfy the read */
-        }
-      })
+      this._sendReadIndexToPeer(peer, idx)
     }
 
     const collabNd =
@@ -664,28 +686,63 @@ export class SwarmDisk {
     }
   }
 
-  async read(index) {
-    if (this.localRAM.has(index)) return this.localRAM.get(index)
-    return new Promise((resolve, reject) => {
-      const timeoutMs = mbrReadTimeoutMsForDisk(this)
-      const timeout = setTimeout(() => {
-        this.pendingReads.delete(index)
-        reject(new Error('MBR read timeout'))
-      }, timeoutMs)
-      this.pendingReads.set(index, (data) => {
-        clearTimeout(timeout)
-        resolve(data)
-      })
-      for (const peer of this.peers) {
-        void peer.chan.fullyOpened().then(() => {
+  /**
+   * Protomux message 0: request block index (MBR). Only send after channel pairing.
+   * @param {unknown} peer
+   * @param {number} index
+   */
+  _sendReadIndexToPeer(peer, index) {
+    const chan = /** @type {{ chan: { messages: { send: (idx: number) => void }[] } }} */ (
+      peer
+    ).chan
+    void chan.fullyOpened().then((opened) => {
+      if (!opened) return
       if (!this.pendingReads.has(index)) return
       try {
-            peer.chan.messages[0].send(index)
+        chan.messages[0].send(index)
       } catch {
         /* ignore — channel closed or not ready */
       }
     })
   }
+
+  /**
+   * Re-broadcast a pending block read to every connected peer (lossy links, slow pairing).
+   * @param {number} index
+   */
+  _broadcastPendingReadIndex(index) {
+    for (const p of this.peers) {
+      this._sendReadIndexToPeer(p, index)
+    }
+  }
+
+  async read(index) {
+    if (this.localRAM.has(index)) return this.localRAM.get(index)
+    return new Promise((resolve, reject) => {
+      const timeoutMs = mbrReadTimeoutMsForDisk(this)
+      /** @type {ReturnType<typeof setInterval> | null} */
+      let rebroadcastIv = null
+      const timeout = setTimeout(() => {
+        if (rebroadcastIv) clearInterval(rebroadcastIv)
+        this.pendingReads.delete(index)
+        reject(new Error('MBR read timeout'))
+      }, timeoutMs)
+      this.pendingReads.set(index, (data) => {
+        if (rebroadcastIv) clearInterval(rebroadcastIv)
+        clearTimeout(timeout)
+        resolve(data)
+      })
+      const rbMs = mbrReadRebroadcastMsFromEnv(globalThis.process?.env)
+      this._broadcastPendingReadIndex(index)
+      if (rbMs > 0) {
+        rebroadcastIv = setInterval(() => {
+          if (!this.pendingReads.has(index)) {
+            if (rebroadcastIv) clearInterval(rebroadcastIv)
+            return
+          }
+          this._broadcastPendingReadIndex(index)
+        }, rbMs)
+      }
     })
   }
 
@@ -714,7 +771,15 @@ export class SwarmDisk {
       state.timeout = setTimeout(() => state.finish(), 3000)
       self.pendingSearches.set(id, state)
       for (const peer of peerList) {
+        void peer.chan.fullyOpened().then((opened) => {
+          if (!opened) return
+          if (state.settled) return
+          try {
             peer.chan.messages[3].send({ id, query })
+          } catch {
+            /* ignore */
+          }
+        })
       }
     })
   }