fix: two boot/federation log errors — duplicate-key upserts and peer-pull denial spam

server/lib/README.md

@@ -436,7 +436,7 @@ The barrel `server/lib/index.js` is a machine-checkable enumeration of every pub
 | `socketEventCatalog.js` | Searchable projection of the cached Socket.IO inventory: direction, domain, and runtime-schema coverage. |
 | `apiOperationContracts.js` | Detailed operation metadata for intentionally public APIs. It consumes the canonical route Zod contracts and feeds both public and internal OpenAPI documents. |
 | `apiRegistry.js` | Single source of truth for which PortOS services are externally-callable HTTP APIs (`voice`, `sdapi`). `API_REGISTRY` declares each API's `publicPrefixes` (read/compute-safe surface only) + defaults; `isRegistryPublic(settings, path)` tells `authGate` when an `exposed && !requireAuth` API re-opens its prefix; `resolveApiAccess(settings)` merges persisted `apiAccess` flags for the Settings UI + OpenAPI docs. |
-| `arrayUtils.js` | `shuffle(arr)` — Fisher-Yates shuffle (new array, never mutates). The canonical uniform shuffle — never `arr.sort(() => Math.random() - 0.5)`, which is biased. Shared by `meatspacePostCognitive.js` (Schulte table / mental rotation) and `meatspacePostMemory.js` (memory drill generators). |
+| `arrayUtils.js` | `shuffle(arr)` — Fisher-Yates shuffle (new array, never mutates). The canonical uniform shuffle — never `arr.sort(() => Math.random() - 0.5)`, which is biased. Shared by `meatspacePostCognitive.js` (Schulte table / mental rotation) and `meatspacePostMemory.js` (memory drill generators). `dedupeByKey(items, keyOf, pick?)` — one survivor per key, first-seen order. **Required before any multi-row `INSERT … ON CONFLICT (key) DO UPDATE`**: Postgres refuses the whole statement ("ON CONFLICT DO UPDATE command cannot affect row a second time") when its VALUES list names one conflict key twice, and the rows a batcher joins usually come from something that promises no uniqueness (a disk scan, a peer payload). `DO NOTHING` upserts are exempt. `pick(held, candidate)` defaults to last-seen-wins (what a sequential one-row upsert loop leaves); pass a comparator when the table's conflict rule isn't "latest write" — `memorySync.applyRemoteChanges` keeps the newest `updatedAt` so a peer's payload ordering can't flip a last-writer-wins outcome. Used by `services/mediaAssetIndex/db.js` and `services/memorySync.js`. |
 | `assetRoutePrefixes.js` | Import-free leaf holding the URL prefixes the server owns: `ASSET_ROUTE_PREFIXES` (every `/data/**` static mount) and `SERVER_OWNED_PREFIXES` (what must never reach the SPA fallback, each with the exact `spaPaths` that ARE client routes). `scripts/dev-proxy-drift.test.js` checks the dev proxy's `^/data/` wildcard against the mounts, pins the route-registration order in `server/index.js` (a router added below the terminators is shadowed), and fails if a client route — from `NAV_COMMANDS` or `App.jsx`'s nested `<Route>` tree — is ever added under a server-owned prefix without being declared. |
 | `asyncMutex.js` | Promise-based async mutex. |
 | `concurrencyGate.js` | `createConcurrencyGate(limit)` → `run(fn)` — cap on simultaneous async work for ONE module-scoped budget, released FIFO. Sibling to `mapWithConcurrency.js`, which caps in-flight work *within one array map*; a gate is shared state, so several call sites fanning out at the same remote respect one budget instead of each respecting its own while the host sees the sum. `createMutex` (`asyncMutex.js`) is this with `limit` fixed at 1 — prefer it for mutual exclusion. Note the budget is per-MODULE, not per-host: two modules calling one host each get their own gate. Used by `huggingFaceCatalog.js` (4) and `ollamaRegistryCatalog.js` (16), whose cold catalog-enrichment bursts otherwise arrive at a free public API as a thundering herd — which the Hub answers with an HTTP/2 GOAWAY that surfaces as a bare `fetch failed`. |

server/lib/arrayUtils.js

@@ -17,3 +17,33 @@ export function shuffle(arr) {
   }
   return out;
 }
+
+/**
+ * Collapse items that share a key, keeping ONE survivor per key in first-seen
+ * order. Returns a new array; never mutates the input.
+ *
+ * The reason this is shared rather than inlined: **a multi-row
+ * `INSERT … ON CONFLICT (key) DO UPDATE` must never name the same conflict key
+ * twice.** Postgres refuses the whole statement with "ON CONFLICT DO UPDATE
+ * command cannot affect row a second time" — it will not pick a winner for you,
+ * because applying two updates to one row in one command has no defined order.
+ * So any batching helper that joins N rows into one VALUES list has to collapse
+ * duplicates first, and the rows it batches usually come from somewhere that
+ * makes no uniqueness promise (a disk scan, a peer's payload). `DO NOTHING`
+ * upserts are exempt — Postgres tolerates in-VALUES duplicates there.
+ *
+ * `pick(held, candidate)` chooses the survivor when a key repeats and must
+ * return one of its two arguments. It defaults to last-seen-wins, which is what
+ * a sequential one-row-at-a-time upsert loop would have left behind. Pass a
+ * comparator instead when the table's conflict rule is not "latest write" — a
+ * last-writer-wins store, for example, has to keep the newest timestamp so a
+ * payload's internal ordering can't change the outcome.
+ */
+export function dedupeByKey(items, keyOf, pick = (held, candidate) => candidate) {
+  const byKey = new Map();
+  for (const item of items) {
+    const key = keyOf(item);
+    byKey.set(key, byKey.has(key) ? pick(byKey.get(key), item) : item);
+  }
+  return [...byKey.values()];
+}

server/lib/arrayUtils.test.js

@@ -1,5 +1,5 @@
 import { describe, it, expect, vi } from 'vitest';
-import { shuffle } from './arrayUtils.js';
+import { shuffle, dedupeByKey } from './arrayUtils.js';
 
 describe('shuffle', () => {
   it('returns a new array — never mutates the input', () => {
@@ -45,3 +45,40 @@ describe('shuffle', () => {
     expect(new Set(out).size).toBe(100);
   });
 });
+
+describe('dedupeByKey', () => {
+  it('keeps one item per key, in first-seen order', () => {
+    const out = dedupeByKey(
+      [{ k: 'a', v: 1 }, { k: 'b', v: 2 }, { k: 'a', v: 3 }],
+      (x) => x.k,
+    );
+    expect(out.map((x) => x.k)).toEqual(['a', 'b']);
+  });
+
+  it('defaults to last-seen-wins — what a sequential upsert loop would leave', () => {
+    const out = dedupeByKey([{ k: 'a', v: 1 }, { k: 'a', v: 2 }], (x) => x.k);
+    expect(out).toEqual([{ k: 'a', v: 2 }]);
+  });
+
+  it('honors a `pick` comparator, so a non-latest conflict rule can win', () => {
+    // memorySync's case: the newest `updatedAt` survives regardless of the
+    // order a peer happened to send the duplicates in.
+    const newest = (held, next) => (held.at > next.at ? held : next);
+    const rows = [{ k: 'a', at: 30 }, { k: 'a', at: 10 }, { k: 'a', at: 20 }];
+    expect(dedupeByKey(rows, (x) => x.k, newest)).toEqual([{ k: 'a', at: 30 }]);
+    // Same set, reversed — the winner must not depend on arrival order.
+    expect(dedupeByKey([...rows].reverse(), (x) => x.k, newest)).toEqual([{ k: 'a', at: 30 }]);
+  });
+
+  it('never mutates the input, and passes a unique list through unchanged', () => {
+    const input = [{ k: 'a' }, { k: 'b' }];
+    const out = dedupeByKey(input, (x) => x.k);
+    expect(out).not.toBe(input);
+    expect(out).toEqual(input);
+    expect(input).toHaveLength(2);
+  });
+
+  it('handles an empty list', () => {
+    expect(dedupeByKey([], (x) => x.k)).toEqual([]);
+  });
+});

server/services/mediaAssetIndex/db.js

@@ -26,6 +26,7 @@
 import { readFile } from 'fs/promises';
 import { join } from 'path';
 import { query } from '../../lib/db.js';
+import { dedupeByKey } from '../../lib/arrayUtils.js';
 import { PATHS } from '../../lib/fileUtils.js';
 import { imageToRow, videoToRow } from './logic.js';
 
@@ -58,8 +59,16 @@ export async function upsertAsset(row) {
 // boot over the whole gallery) is a handful of round-trips, not one-per-asset.
 const UPSERT_CHUNK = 500;
 async function upsertAssets(rows) {
-  for (let i = 0; i < rows.length; i += UPSERT_CHUNK) {
-    const chunk = rows.slice(i, i + UPSERT_CHUNK);
+  // Collapse duplicate media_keys BEFORE chunking (see `dedupeByKey` for why a
+  // multi-row upsert cannot carry a repeated conflict key). Untreated, one throw
+  // aborted the WHOLE reconcile — a single duplicated gallery filename or
+  // repeated video-history id froze the entire index and turned boot's
+  // catalog-migration step red. Disk is the authority and it can honestly hand
+  // us the same ref twice (a re-appended history entry, a filename two scans
+  // both list); that is data to absorb, not an error to propagate.
+  const deduped = dedupeByKey(rows, (row) => row.mediaKey);
+  for (let i = 0; i < deduped.length; i += UPSERT_CHUNK) {
+    const chunk = deduped.slice(i, i + UPSERT_CHUNK);
     const values = [];
     const params = [];
     chunk.forEach((row, j) => {
@@ -75,6 +84,7 @@ async function upsertAssets(rows) {
       params,
     );
   }
+  return deduped.length;
 }
 
 /** Remove one index row by media_key. */
@@ -182,7 +192,8 @@ export async function reconcileMediaAssets(deps = {}) {
   const videoRows = (Array.isArray(videoRead.list) ? videoRead.list : [])
     .map((v) => videoToRow(v, { now })).filter(Boolean);
 
-  await upsertAssets([...imageRows, ...videoRows]);
+  // Rows WRITTEN, which is below the rows read when disk repeated a ref.
+  const indexed = await upsertAssets([...imageRows, ...videoRows]);
 
   // Per-kind prune, gated on a successful read for that kind. Pruning one kind
   // never touches the other's rows (an image-read failure can't wipe videos).
@@ -193,7 +204,7 @@ export async function reconcileMediaAssets(deps = {}) {
   const skipped = [!imageRead.ok && 'images', !videoRead.ok && 'videos'].filter(Boolean);
   const skipNote = skipped.length ? ` — SKIPPED prune for ${skipped.join('+')} (disk read failed)` : '';
   console.log(`🗂️  Media asset index reconciled: ${imageRows.length} img / ${videoRows.length} vid on disk, ${pruned} stale row(s) pruned${skipNote}`);
-  return { ok: true, indexed: imageRows.length + videoRows.length, pruned, skippedPrune: skipped };
+  return { ok: true, indexed, pruned, skippedPrune: skipped };
 }
 
 // Delete index rows of `kind` whose media_key isn't in `liveKeys`. An empty

server/services/mediaAssetIndex/db.test.js

@@ -149,6 +149,33 @@ describe.skipIf(!runDb)('media asset index DB round-trip', () => {
     expect(vids.some((x) => x.id === `${PFX}vid1`)).toBe(true);
   });
 
+  it('survives duplicate refs on disk (multi-row upsert must not self-conflict)', async () => {
+    // Catches the multi-row upsert self-conflicting on a repeated media_key —
+    // a Postgres-level constraint no unit test can pin. See the dedupe comment
+    // in db.js for why disk can hand us the same ref twice.
+    const listGallery = async () => [
+      { filename: `${PFX}dup.png`, prompt: 'first', createdAt: '2026-04-01T00:00:00.000Z' },
+      { filename: `${PFX}dup.png`, prompt: 'second', createdAt: '2026-04-02T00:00:00.000Z' },
+    ];
+    const loadHistory = async () => [
+      { id: `${PFX}dupvid`, filename: `${PFX}dupvid.mp4`, createdAt: '2026-04-03T00:00:00.000Z' },
+      { id: `${PFX}dupvid`, filename: `${PFX}dupvid.mp4`, createdAt: '2026-04-04T00:00:00.000Z' },
+    ];
+
+    const res = await db.reconcileMediaAssets({ listGallery, loadHistory });
+    // `indexed` counts rows written, so the collapsed pair counts once each.
+    expect(res.indexed).toBe(2);
+
+    // Last occurrence wins — the same row a sequential upsert loop would leave.
+    const imgs = await db.listAssets({ kind: 'image' });
+    const dup = imgs.filter((x) => x.filename === `${PFX}dup.png`);
+    expect(dup).toHaveLength(1);
+    expect(dup[0].prompt).toBe('second');
+
+    const vids = await db.listAssets({ kind: 'video' });
+    expect(vids.filter((x) => x.id === `${PFX}dupvid`)).toHaveLength(1);
+  });
+
   it('does NOT prune a kind whose disk read failed — skips, keeps live rows', async () => {
     // Seed an image row that a healthy reconcile would normally prune (its file
     // is not in the "disk" set), and a video row that the healthy video read

server/services/memorySync.db.test.js

@@ -0,0 +1,155 @@
+/**
+ * Postgres-backed tests for `applyRemoteChanges` — the federation write path.
+ *
+ * This is the one function in `memorySync.js` that takes a REMOTE peer's payload
+ * and writes it, so it is the place where an assumption about that payload's
+ * shape becomes a data bug. It batches 100 rows into a single multi-row
+ * `INSERT … ON CONFLICT (id) DO UPDATE` inside one transaction, which makes two
+ * properties worth pinning against a real Postgres:
+ *
+ *   - a repeated id in the payload must not abort the apply (Postgres refuses a
+ *     multi-row upsert that names one conflict key twice), and
+ *   - the last-writer-wins rule must be decided by `updated_at`, never by where
+ *     a row happened to sit in the peer's list.
+ *
+ * Neither is expressible without the database: the first is a Postgres statement
+ * constraint and the second lives in the `WHERE EXCLUDED.updated_at >` clause.
+ *
+ * `*.db.test.js` → runs ONLY via `npm run test:db` against `portos_test`, never
+ * the real `portos` DB (the db.js runner guard + the suite skip below enforce
+ * this). Registered in `DB_TEST_INCLUDE` in `vitest.config.db.js`, without which
+ * it would silently never run.
+ *
+ * `applyRemoteChanges` reads and writes the whole `memories` table by id, so
+ * like `memoryDB.db.test.js` this suite clears the table and seeds its own rows.
+ * `fileParallelism: false` means nothing else is touching it meanwhile.
+ */
+
+import { describe, it, expect, beforeAll, afterAll, beforeEach } from 'vitest';
+import { checkHealth, ensureSchema, close, query } from '../lib/db.js';
+import { requireDbOrSkip } from '../lib/dbTestGate.js';
+
+let dbReady = false;
+let skipReason = '';
+{
+  const health = await checkHealth().catch((e) => ({ connected: false, error: e?.message }));
+  if (!health.connected) {
+    skipReason = `Postgres not reachable (${health.error || 'no connection'})`;
+  } else {
+    await ensureSchema().catch(() => {});
+    const recheck = await checkHealth().catch(() => ({ hasSchema: false }));
+    if (recheck.hasSchema) dbReady = true;
+    else skipReason = 'memory schema not present';
+  }
+}
+const runDb = requireDbOrSkip('services/memorySync.db.test', dbReady, skipReason);
+
+// Obviously-fake ids; the payload shape is what applyRemoteChanges expects off
+// the wire (camelCase, since it maps to columns itself).
+const ID_A = '00000000-0000-4000-8000-00000000aaaa';
+const ID_B = '00000000-0000-4000-8000-00000000bbbb';
+
+const remoteMemory = (id, updatedAt, overrides = {}) => ({
+  id,
+  type: 'fact',
+  content: `content @ ${updatedAt}`,
+  summary: null,
+  category: 'general',
+  tags: [],
+  embedding: null,
+  embeddingModel: null,
+  confidence: 0.5,
+  importance: 0.5,
+  status: 'active',
+  sourceTaskId: null,
+  sourceAgentId: null,
+  sourceAppId: null,
+  expiresAt: null,
+  createdAt: '2026-01-01T00:00:00.000Z',
+  updatedAt,
+  originInstanceId: '00000000-0000-4000-8000-00000000feed',
+  ...overrides,
+});
+
+const contentOf = async (id) => {
+  const res = await query(`SELECT content FROM memories WHERE id = $1`, [id]);
+  return res.rows[0]?.content ?? null;
+};
+
+describe.skipIf(!runDb)('memorySync.applyRemoteChanges', () => {
+  let memorySync;
+  beforeAll(async () => {
+    memorySync = await import('./memorySync.js');
+  });
+  beforeEach(async () => {
+    await query(`DELETE FROM memories`);
+  });
+  afterAll(async () => {
+    await query(`DELETE FROM memories`).catch(() => {});
+    await close();
+  });
+
+  it('applies a duplicated id instead of aborting the whole transaction', async () => {
+    // Regression: a peer payload repeating one id made the batched multi-row
+    // upsert throw "ON CONFLICT DO UPDATE command cannot affect row a second
+    // time". Because the batches share a transaction, that rolled back EVERY
+    // row in the payload — including the ones with no duplicate at all, which is
+    // what turns a peer's malformed page into a permanently stalled sync.
+    const result = await memorySync.applyRemoteChanges([
+      remoteMemory(ID_A, '2026-05-01T00:00:00.000Z'),
+      remoteMemory(ID_A, '2026-05-02T00:00:00.000Z'),
+      remoteMemory(ID_B, '2026-05-01T00:00:00.000Z'),
+    ]);
+
+    // The unrelated row survived — that is the property the rollback destroyed.
+    expect(await contentOf(ID_B)).toBe('content @ 2026-05-01T00:00:00.000Z');
+    expect(await contentOf(ID_A)).toBe('content @ 2026-05-02T00:00:00.000Z');
+    // The collapsed copy lost last-writer-wins, so it counts as skipped and the
+    // three tallies still account for every row the peer sent.
+    expect(result).toEqual({ inserted: 2, updated: 0, skipped: 1 });
+  });
+
+  it('resolves a duplicated id by updated_at, not by payload order', async () => {
+    // The tie-break has to match what the ON CONFLICT clause would have done had
+    // the same two rows arrived in separate batches — otherwise a peer could
+    // flip which copy wins just by reordering its page.
+    await memorySync.applyRemoteChanges([
+      remoteMemory(ID_A, '2026-06-09T00:00:00.000Z'),   // newest, sent FIRST
+      remoteMemory(ID_A, '2026-06-01T00:00:00.000Z'),
+    ]);
+    expect(await contentOf(ID_A)).toBe('content @ 2026-06-09T00:00:00.000Z');
+  });
+
+  it('keeps the parseable copy when a duplicate carries a malformed clock', async () => {
+    // A malformed `updatedAt` must not win the collapse: it would be handed to a
+    // timestamptz column and fail the statement, losing a row we could have
+    // applied intact. NaN compares false against everything, so the comparator
+    // has to sort it below a real clock explicitly rather than by accident.
+    const result = await memorySync.applyRemoteChanges([
+      remoteMemory(ID_A, '2026-09-01T00:00:00.000Z'),
+      remoteMemory(ID_A, 'not-a-timestamp'),
+    ]);
+    expect(await contentOf(ID_A)).toBe('content @ 2026-09-01T00:00:00.000Z');
+    expect(result).toMatchObject({ inserted: 1, skipped: 1 });
+  });
+
+  it('still refuses a remote row older than the local one (last-writer-wins)', async () => {
+    await memorySync.applyRemoteChanges([remoteMemory(ID_A, '2026-07-10T00:00:00.000Z')]);
+
+    const result = await memorySync.applyRemoteChanges([
+      remoteMemory(ID_A, '2026-07-01T00:00:00.000Z'),
+    ]);
+    expect(await contentOf(ID_A)).toBe('content @ 2026-07-10T00:00:00.000Z');
+    expect(result).toMatchObject({ inserted: 0, updated: 0, skipped: 1 });
+  });
+
+  it('updates in place when the remote row is newer', async () => {
+    await memorySync.applyRemoteChanges([remoteMemory(ID_A, '2026-08-01T00:00:00.000Z')]);
+
+    const result = await memorySync.applyRemoteChanges([
+      remoteMemory(ID_A, '2026-08-20T00:00:00.000Z'),
+    ]);
+    expect(await contentOf(ID_A)).toBe('content @ 2026-08-20T00:00:00.000Z');
+    expect(result).toMatchObject({ inserted: 0, updated: 1, skipped: 0 });
+  });
+});