feat(host-store): antseed file state + call-ledger route identity off the filesystem

Dockerfile.antseed

@@ -10,9 +10,12 @@ RUN apt-get update \
     && rm -rf /var/lib/apt/lists/*
 
 # Pinned, vendored. Bump deliberately (newer CLIs carry fresher bootstrap nodes).
-RUN npm install -g @antseed/cli@0.1.128
+# `pg` is the market writer's Postgres client (write-market.js upserts peer_offers
+# into the shared host store). NODE_PATH lets the scripts require these globals.
+RUN npm install -g @antseed/cli@0.1.128 pg@8.16.3
+ENV NODE_PATH=/usr/local/lib/node_modules
 
-COPY antseed/merge-market.js antseed/control.js antseed/entrypoint.sh /usr/local/lib/antseed/
+COPY antseed/store.js antseed/write-market.js antseed/write-status.js antseed/control.js antseed/entrypoint.sh /usr/local/lib/antseed/
 RUN chmod +x /usr/local/lib/antseed/entrypoint.sh
 
 ENV ANTSEED_DATA_DIR=/data

antseed/control.js

@@ -12,15 +12,19 @@
 'use strict';
 const http = require('http');
 const { execFile } = require('child_process');
-const fs = require('fs');
+const { Pool } = require('pg');
+const { UPSERT_BUYER_STATUS, buyerStatusRow } = require('./store.js');
 
 const PORT = parseInt(process.env.ANTSEED_CONTROL_PORT || '8379', 10);
 const TOKEN = process.env.ANTSEED_CONTROL_TOKEN || '';
-const MARKET_DIR = process.env.ANTSEED_MARKET_DIR || '/market';
-const STATUS_FILE = MARKET_DIR + '/status-antseed.json';
+const PID = process.env.ANTSEED_BUYER_PID || 'antseed';
 const DEPOSIT_TIMEOUT_MS = 120000; // on-chain tx
 const STATUS_TIMEOUT_MS = 30000;
 
+// One pool for the long-lived control server (write-status.js, the poll-loop
+// twin, is one-shot and uses a plain Client instead).
+const pool = new Pool({ connectionString: process.env.DATABASE_URL });
+
 if (!TOKEN) {
   console.error('[control] ANTSEED_CONTROL_TOKEN unset — control server disabled');
   return;
@@ -42,20 +46,22 @@ function run(args, timeout) {
   });
 }
 
-// Refresh /market/status-antseed.json from a fresh `buyer status --json`, in the
-// same shape the entrypoint writer uses (fetched_at_ms + atomic rename) so the
-// router's source picks up the new escrow balance on its next read.
+// Upsert the buyer's status into the host store (buyer_status) from a fresh
+// `buyer status --json`, so the router's source picks up the new escrow balance
+// on its next read — the post-wallet-op twin of write-status.js. Returns the
+// fresh status object for the HTTP response even if the persist fails (the poll
+// loop will retry the write); null only when the CLI output isn't a status.
 async function refreshStatus() {
   const r = await run(['buyer', 'status', '--json'], STATUS_TIMEOUT_MS);
   let data;
   try { data = JSON.parse(r.stdout); } catch (_) { return null; }
   if (data === null || typeof data !== 'object') return null;
   data.fetched_at_ms = Date.now();
-  const tmp = STATUS_FILE + '.' + process.pid + '.tmp';
   try {
-    fs.writeFileSync(tmp, JSON.stringify(data));
-    fs.renameSync(tmp, STATUS_FILE);
-  } catch (_) { try { fs.unlinkSync(tmp); } catch (__) {} }
+    await pool.query(UPSERT_BUYER_STATUS, buyerStatusRow(data, PID));
+  } catch (e) {
+    console.error('[control] buyer_status upsert failed:', e.message);
+  }
   return data;
 }
 

antseed/entrypoint.sh

@@ -10,32 +10,44 @@
 #     socat exposes it on the container network. This lets us DROP
 #     `network_mode: service:router`, whose orphaned netns (router recreated,
 #     antseed not) silently zeroed discovery on every deploy.
-#   * validate-before-write      -> never clobber market.json with the CLI's
-#     non-JSON "No peers found" line.
+#   * validate-before-write      -> the writers reject the CLI's non-JSON
+#     "No peers found" line (exit non-zero) instead of wiping the store row.
 #   * --top high                 -> the CLI default is 20; we want the whole
 #     peer set.
+#
+# The market book and buyer status are written to the host store (Postgres)
+# by write-market.js / write-status.js — no market.json / status-*.json files.
+# $MARKET_DIR is now just container-local scratch for the raw CLI dumps.
 set -eu
 
 PORT_PROXY="${ANTSEED_PROXY_PORT:-8377}"      # buyer proxy (binds 127.0.0.1)
 PORT_PUBLIC="${ANTSEED_PUBLIC_PORT:-8378}"    # socat, on the container network
 MARKET_DIR="${ANTSEED_MARKET_DIR:-/market}"
 TOP="${ANTSEED_BROWSE_TOP:-500}"
 INTERVAL="${ANTSEED_BROWSE_INTERVAL:-60}"     # browse cadence (s); 60s feeds the
-                                              # sliding window in merge-market.js
+                                              # observed_at sliding window in peer_offers
 MAXIN="${ANTSEED_MAX_INPUT:-1000}"            # buyer spend rail; Σ_pol policy is
 MAXOUT="${ANTSEED_MAX_OUTPUT:-1000}"          # the real per-call price ceiling
 CLI_TIMEOUT="${ANTSEED_CLI_TIMEOUT:-45}"      # hard cap per browse/status CLI call.
                                               # WITHOUT it a single hung `antseed`
                                               # invocation (e.g. a concurrent buyer
                                               # command grabbing the sqlite store)
                                               # blocks the writer loop FOREVER — it
-                                              # has no self-restart — so market.json
-                                              # and status-antseed.json silently
-                                              # freeze and the catalog/wallet go stale.
+                                              # has no self-restart — so the
+                                              # peer_offers / buyer_status rows
+                                              # silently freeze and the catalog/wallet go stale.
 LIB=/usr/local/lib/antseed
 
 mkdir -p "$MARKET_DIR"
 
+# A CHANGE_ME / non-hex ANTSEED_IDENTITY_HEX (the unset-secret placeholder) would
+# make the CLI reject the identity; unset it so the buyer falls back to a
+# generated key on the data volume — durable enough for dev, and the prod secret
+# is a real 64-hex hot-wallet.
+if ! printf '%s' "${ANTSEED_IDENTITY_HEX:-}" | grep -qiE '^[0-9a-f]{64}$'; then
+    unset ANTSEED_IDENTITY_HEX
+fi
+
 # Buyer proxy in browse mode (no --peer: the host pins per request). A funded
 # wallet is needed to transact; discovery/pricing work unfunded.
 antseed buyer start -p "$PORT_PROXY" \
@@ -50,24 +62,12 @@ socat "TCP-LISTEN:${PORT_PUBLIC},fork,reuseaddr" "TCP:127.0.0.1:${PORT_PROXY}" &
 # Self-disables when ANTSEED_CONTROL_TOKEN is unset. See antseed/control.js.
 node "$LIB/control.js" &
 
-atomic_write() {  # <validator-cmd> <dest> ; reads producer stdout on fd via $1
-    dest="$1"; tmp="${dest}.$$.tmp"
-    if cat > "$tmp" && [ -s "$tmp" ]; then
-        mv "$tmp" "$dest"
-    else
-        rm -f "$tmp"
-        return 1
-    fi
-}
-
 write_market() {
     raw="$MARKET_DIR/.market.raw.$$"
     timeout -k 5 "$CLI_TIMEOUT" antseed network browse --services --top "$TOP" --json > "$raw" 2>/dev/null || true
-    # merge this browse into the rolling window (and validate: a non-dump exits
-    # non-zero -> keep the last good market.json)
-    if node "$LIB/merge-market.js" "$MARKET_DIR/market.json" < "$raw" | atomic_write "$MARKET_DIR/market.json"; then
-        :
-    else
+    # upsert this browse into the host store's peer_offers (and validate: a
+    # non-dump / DB error exits non-zero -> keep the last good window, no write)
+    if ! node "$LIB/write-market.js" < "$raw"; then
         # record what we got for debugging; keep the last good window
         cp "$raw" "$MARKET_DIR/market.err" 2>/dev/null || true
     fi
@@ -77,8 +77,9 @@ write_market() {
 write_status() {
     raw="$MARKET_DIR/.status.raw.$$"
     timeout -k 5 "$CLI_TIMEOUT" antseed buyer status --json > "$raw" 2>/dev/null || true
-    node -e 'const fs=require("fs");let d;try{d=JSON.parse(fs.readFileSync(0,"utf8"))}catch(e){process.exit(2)}if(d===null||typeof d!=="object")process.exit(3);d.fetched_at_ms=Date.now();process.stdout.write(JSON.stringify(d))' \
-        < "$raw" | atomic_write "$MARKET_DIR/status-antseed.json" || true
+    # upsert the buyer status into the host store's buyer_status (validate + DB
+    # error -> exit non-zero, keep the last good row)
+    node "$LIB/write-status.js" < "$raw" || true
     rm -f "$raw"
 }
 

antseed/store.js

@@ -0,0 +1,27 @@
+// Shared host-store writes for the antseed sidecar (write-status.js on the poll
+// loop and control.js after a wallet op both upsert the buyer status). One place
+// for the buyer_status row shape + UPSERT so the two writers can't drift.
+//
+// Raw buyer-reported fields as columns; the deposits are kept as the strings the
+// buyer reports (sources/antseed coerces on read). No interpretation here.
+const UPSERT_BUYER_STATUS = `INSERT INTO buyer_status
+  (pid, pinned_peer_id, deposits_available, deposits_reserved, wallet_address,
+   connection_state, fetched_at)
+  VALUES ($1,$2,$3,$4,$5,$6,$7)
+  ON CONFLICT (pid) DO UPDATE SET
+    pinned_peer_id=EXCLUDED.pinned_peer_id,
+    deposits_available=EXCLUDED.deposits_available,
+    deposits_reserved=EXCLUDED.deposits_reserved,
+    wallet_address=EXCLUDED.wallet_address,
+    connection_state=EXCLUDED.connection_state,
+    fetched_at=EXCLUDED.fetched_at`;
+
+const str = (v) => (v === null || v === undefined) ? null : String(v);
+
+function buyerStatusRow(d, pid) {
+  return [pid, str(d.pinnedPeerId), str(d.depositsAvailable),
+          str(d.depositsReserved), str(d.walletAddress),
+          str(d.connectionState), Date.now()];
+}
+
+module.exports = { UPSERT_BUYER_STATUS, buyerStatusRow };

antseed/write-market.js

@@ -0,0 +1,85 @@
+// Marketplace book writer (replaces merge-market.js). Each browse sees only the
+// peers DHT-reachable from this node *right now* (~68); the network has more
+// (~83) that flit in and out of our view. We keep a SLIDING window so a peer
+// silent for a few browses still routes: instead of a hand-rolled union of the
+// previous file, every peer/service we see is UPSERT-ed into the host store's
+// `peer_offers` table stamped `observed_at = now`, and sources/antseed.py reads
+// only rows within the window (WHERE observed_at >= now - window). The window
+// thus lives as a read-time filter; here we just record what we saw and prune
+// rows older than the window so the table can't grow without bound.
+//
+// One RAW row per (peerId, service): the seller's announced prices/cap/reputation
+// as columns, no interpretation — ranking/admission stays in offers_sync (host)
+// and the Σ_pol policy (core). Type-cleaning (positive-int cap, numeric
+// reputation, non-null cached price) happens here, at the write, mirroring the
+// coercion sources/antseed.py used to do at the read.
+//
+// Validation: a non-dump (the CLI prints a human "No peers found" line even with
+// --json) exits non-zero so entrypoint.sh keeps the last good window (no write).
+const { Client } = require("pg");
+
+const WINDOW_MS = Number(process.env.ANTSEED_PEER_WINDOW_MS || 15 * 60 * 1000);
+const now = Date.now();
+
+function numOr0(v) { const n = Number(v); return Number.isFinite(n) ? n : 0; }
+function numOrNull(v) {
+  if (v === null || v === undefined) return null;
+  const n = Number(v);
+  return Number.isFinite(n) ? n : null;
+}
+function posIntOrNull(v) {  // maxConcurrency: positive int, else ungated (null)
+  const n = Number(v);
+  return Number.isFinite(n) && n > 0 ? Math.trunc(n) : null;
+}
+
+let fresh;
+try { fresh = JSON.parse(require("fs").readFileSync(0, "utf8")); }
+catch (e) { process.exit(2); }                            // not JSON
+if (fresh === null || typeof fresh !== "object" || !Array.isArray(fresh.peers))
+  process.exit(3);                                        // JSON but not a dump
+
+// Flatten the nested browse dump (peer -> providerPricing -> services) to one
+// row per (peerId, service). No services -> the peer contributes no rows.
+const rows = [];
+for (const peer of fresh.peers) {
+  if (!peer || !peer.peerId) continue;
+  const maxc = posIntOrNull(peer.maxConcurrency);
+  const rep = numOrNull(peer.onChainReputationScore);
+  const lastSeen = numOrNull(peer.lastSeen);
+  for (const pricing of Object.values(peer.providerPricing || {})) {
+    for (const [service, sp] of Object.entries((pricing || {}).services || {})) {
+      rows.push([
+        peer.peerId, service,
+        numOr0(sp.inputUsdPerMillion), numOr0(sp.outputUsdPerMillion),
+        numOrNull(sp.cachedInputUsdPerMillion),
+        maxc, rep, lastSeen, now, now, now,
+      ]);
+    }
+  }
+}
+
+const UPSERT = `INSERT INTO peer_offers
+  (peer_id, service, price_in, price_out, price_cached_in, max_concurrency,
+   reputation, last_seen, observed_at, first_seen, fetched_at)
+  VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11)
+  ON CONFLICT (peer_id, service) DO UPDATE SET
+    price_in=EXCLUDED.price_in, price_out=EXCLUDED.price_out,
+    price_cached_in=EXCLUDED.price_cached_in,
+    max_concurrency=EXCLUDED.max_concurrency, reputation=EXCLUDED.reputation,
+    last_seen=EXCLUDED.last_seen, observed_at=EXCLUDED.observed_at,
+    fetched_at=EXCLUDED.fetched_at`;  // first_seen preserved across conflicts
+
+(async () => {
+  const client = new Client({ connectionString: process.env.DATABASE_URL });
+  try {
+    await client.connect();
+    for (const r of rows) await client.query(UPSERT, r);
+    await client.query("DELETE FROM peer_offers WHERE observed_at < $1",
+                       [now - WINDOW_MS]);
+  } catch (e) {
+    process.stderr.write(`write-market: ${e.message}\n`);
+    process.exitCode = 4;                                 // DB error -> keep window
+  } finally {
+    await client.end().catch(() => {});
+  }
+})();

antseed/write-status.js

@@ -0,0 +1,29 @@
+// Buyer status writer (replaces the inline `node -e` + status-antseed.json).
+// Reads `antseed buyer status --json` from stdin and UPSERTs the buyer's status
+// (session pin + escrow + wallet) into the host store's buyer_status, keyed by
+// the buyer pid. sources/antseed.py reads it (_pinned_peer + balances).
+//
+// Validation: a non-object / non-JSON exits non-zero so entrypoint.sh keeps the
+// last good row (no write).
+const { Client } = require("pg");
+const { UPSERT_BUYER_STATUS, buyerStatusRow } = require("./store.js");
+
+const PID = process.env.ANTSEED_BUYER_PID || "antseed";
+
+let d;
+try { d = JSON.parse(require("fs").readFileSync(0, "utf8")); }
+catch (e) { process.exit(2); }                            // not JSON
+if (d === null || typeof d !== "object") process.exit(3); // JSON but not a status
+
+(async () => {
+  const client = new Client({ connectionString: process.env.DATABASE_URL });
+  try {
+    await client.connect();
+    await client.query(UPSERT_BUYER_STATUS, buyerStatusRow(d, PID));
+  } catch (e) {
+    process.stderr.write(`write-status: ${e.message}\n`);
+    process.exitCode = 4;                                 // DB error -> keep row
+  } finally {
+    await client.end().catch(() => {});
+  }
+})();