diff --git a/scripts/drivers/storage/sqlite-sync.sh b/scripts/drivers/storage/sqlite-sync.sh index 3027e7c79..6647529fd 100644 --- a/scripts/drivers/storage/sqlite-sync.sh +++ b/scripts/drivers/storage/sqlite-sync.sh @@ -766,7 +766,9 @@ storage_sync_reconcile_push() { done [ "$count" -gt 0 ] || return 13 - agmsg_sqlite "$db" "BEGIN IMMEDIATE; + # Stdin, for the same reason as the pull outcomes (#882): `$values` gains an + # entry per acked message and a full catch-up push carries a thousand. + printf '%s\n' "BEGIN IMMEDIATE; CREATE TEMP TABLE incoming_sync_acks( local_position INTEGER UNIQUE,wire_id TEXT UNIQUE,server_seq TEXT UNIQUE); INSERT INTO incoming_sync_acks VALUES $values; @@ -807,7 +809,7 @@ storage_sync_reconcile_push() { WHERE b.local_team='$tl' AND b.server_instance_id='$server' AND b.remote_team_id='$remote' AND b.protocol_version=$protocol AND b.driver_generation='$generation'; - COMMIT;" >/dev/null 2>&1 || return 12 + COMMIT;" | agmsg_sqlite -batch "$db" >/dev/null 2>&1 || return 12 _sqlite_data "$team" "SELECT json_object('type','sync_reconcile_result','push_cursor', CAST(push_cursor AS TEXT)) FROM sync_bindings WHERE local_team='$tl' @@ -1091,7 +1093,11 @@ storage_sync_apply_pull() { AND status='corrupt_state') + (SELECT COUNT(*) FROM sync_conflicts WHERE server_instance_id='$server' AND remote_team_id='$remote' AND protocol_version=$protocol);" | tr -d '\r') - _sqlite_data "$team" "SELECT json_object('type','sync_apply_result','transport_cursor', + # STDIN, BECAUSE THIS ONE GROWS WITH THE PAGE (#882). `outcome_ids` gains an + # entry per pulled message and is embedded TWICE below, so the command line + # this used to be would pass about 400 messages on Windows and refuse the + # next one. Nothing else about the query changed. + _sqlite_data_stdin "$team" "SELECT json_object('type','sync_apply_result','transport_cursor', transport_cursor,'corrupt_count',$corrupt) FROM sync_bindings WHERE local_team='$tl' AND server_instance_id='$server' AND remote_team_id='$remote' AND protocol_version=$protocol AND driver_generation='$generation'; @@ -1225,7 +1231,9 @@ EOF insert_local_agents="INSERT INTO local_read_agents VALUES $local_values;" fi - agmsg_sqlite "$db" "BEGIN IMMEDIATE; + # Stdin, third of the same kind (#882): `$insert_members` carries one row per + # roster member and `$insert_local_agents` one per local agent. + printf '%s\n' "BEGIN IMMEDIATE; CREATE TEMP TABLE incoming_read_members(member_id TEXT UNIQUE,agent TEXT UNIQUE); CREATE TEMP TABLE local_read_agents(agent TEXT PRIMARY KEY); $insert_members @@ -1320,7 +1328,7 @@ EOF AND rm.remote_team_id='$remote' AND rm.protocol_version=$protocol AND rm.driver_generation='$generation' AND rm.active=1 AND rm.name_mismatch=0; - COMMIT;" >/dev/null || return 13 + COMMIT;" | agmsg_sqlite -batch "$db" >/dev/null || return 13 _sqlite_data "$team" "SELECT json_object('type','sync_read_frontier','member_id',f.member_id, 'server_seq',f.server_seq) FROM sync_read_prepared f JOIN sync_read_members rm diff --git a/scripts/drivers/storage/sqlite.sh b/scripts/drivers/storage/sqlite.sh index a3ee3e46b..4ede31876 100755 --- a/scripts/drivers/storage/sqlite.sh +++ b/scripts/drivers/storage/sqlite.sh @@ -34,6 +34,28 @@ _sqlite_data() { ( set -o pipefail; agmsg_sqlite "$(_sqlite_db "$1")" "$2" | tr -d '\r' ) } +# The same query, handed over stdin instead of on the command line (#882). +# +# FOR SQL WHOSE LENGTH GROWS WITH THE DATA, and only for that. A command line +# has an operating-system limit and stdin does not, so any statement carrying a +# list of ids -- one `IN (...)` entry per pulled message, per acked message, per +# roster member -- has to arrive this way or it stops working at a size nobody +# chose. +# +# The size that stops it is not large. Windows' CreateProcess caps the command +# line at 32,767 characters; measured on a Windows machine, sqlite3 took 827 +# uuids as arguments and refused 837. A pull page carrying its ids twice +# reaches that at about 400 messages, which is under half a default page, so a +# team that had grown past it simply could not be pulled -- the failure the +# report in #882 arrived as. +# +# `-batch` because this is a script rather than a session: without it sqlite3 +# reading a non-tty is still willing to treat a malformed line as an +# interactive prompt, and the point of this path is that nobody is watching. +_sqlite_data_stdin() { + ( set -o pipefail; printf '%s\n' "$2" | agmsg_sqlite -batch "$(_sqlite_db "$1")" | tr -d '\r' ) +} + # IN (...) list of "team:agent" pairs. _sqlite_pair_in() { local out="" p t a diff --git a/scripts/internal/remote-sync.mjs b/scripts/internal/remote-sync.mjs index 893650b5e..681421cc2 100755 --- a/scripts/internal/remote-sync.mjs +++ b/scripts/internal/remote-sync.mjs @@ -3435,6 +3435,35 @@ export async function resyncCycle(config, acceptedFloor, dependencies = {}) { // credential. What it shares with the normal pull is the part that must not // diverge -- evaluatePull decides what may be imported, then roster mutations // and chat records go through the same two drivers as a connected cycle. +/** + * A line of progress for a command that is otherwise silent for minutes (#882). + * + * STDERR, NEVER STDOUT. `cmd_pull` captures this process's stdout as the result + * channel -- `result="$(... pull-bootstrap ...)"` -- so anything written there + * is invisible until the command finishes AND rides in the same stream the + * caller greps for `pull_bootstrap_result`. stderr is not redirected by that + * caller, so it reaches the operator live and changes no parsing. + * + * WHY IT EXISTS: a verifier ran this against a real server, saw nothing for + * five minutes, concluded it had hung, and killed it. On Linux the same command + * takes 312 seconds and succeeds. Silence and a stall were indistinguishable, + * so thirty minutes went into diagnosing a command that was working. + * + * The lines separate FETCHING from APPLYING on purpose. Applying spawns a child + * process per batch; fetching does not. When this stops moving, which line it + * stopped on says which half to look at -- and on Windows, where the report + * came from, that is the whole question. + */ +function pullProgress(startedAt, message) { + // Elapsed, because a terminal does not timestamp its own scrollback. Without + // it a report of "it printed this and then stopped" cannot say whether the + // stop was ten seconds or ten minutes -- and on the Linux run that finished, + // the whole thing took 312 seconds, so a slow phase and a stuck one look the + // same to whoever is watching. + const elapsed = Math.round((Date.now() - startedAt) / 1000); + process.stderr.write(`agmsg: [${elapsed}s] ${message}\n`); +} + export async function pullBootstrap(args, dependencies = {}) { const publicSnapshotCall = dependencies.publicSnapshotCall ?? publicSnapshot; const requestPublicCall = dependencies.requestPublicCall ?? requestPublic; @@ -3452,6 +3481,25 @@ export async function pullBootstrap(args, dependencies = {}) { // There is no connected binding yet, so this one call validates itself // rather than going through the checks the rest of the pull relies on. + const pullStartedAt = Date.now(); + // THE HOST, NEVER THE ENDPOINT. A hosted endpoint is `https://host/t/` + // and that token IS the capability: anyone who reads it off a terminal, a + // screen share, or a pasted log can connect as this team. This line is the + // one people paste -- it exists because someone sat in front of a silent + // command for 79 minutes and then wrote an issue about it. + // + // `hostOf` rather than anything written here: it is `new URL(...).host`, which + // drops path, query, fragment AND userinfo, and is already the rule this file + // uses for the refusal record `status` prints. `remote.sh` holds the same rule + // in shell (`_remote_endpoint_display`), and the reason both drop the path + // before the userinfo is that an `@` inside a path would otherwise decide + // where the host ends -- a URL parser gets that right without being told. + // + // The team name is what makes host-only enough to name the destination: a + // team has one endpoint. + pullProgress(pullStartedAt, + `pulling ${team} from ${hostOf(serverUrl) ?? "an unreadable endpoint"}` + + " -- this can take several minutes"); const teamSnapshot = await publicSnapshotCall(serverUrl, teamId); const config = { format_version: 1, @@ -3475,8 +3523,16 @@ export async function pullBootstrap(args, dependencies = {}) { let imported = 0; let ageV1Envelopes = 0; for (;;) { + // The cursor is the SERVER's value, and this path does not put it through + // `sequence()` before using it. A canonical sequence is digits; anything + // else goes to a terminal as a placeholder rather than as itself, because + // this line is pasted and control characters travel. + const shownCursor = /^[0-9]{1,20}$/.test(cursor) ? cursor : "an unreadable cursor"; + pullProgress(pullStartedAt, + `fetching messages after ${shownCursor} (${imported} pulled so far)`); const page = await requestPublicCall(config, `/v1/teams/${teamId}/messages?after=${cursor}&limit=${limit}`); + pullProgress(pullStartedAt, `applying ${page.messages.length} messages`); const records = []; for (const message of page.messages) { if (message.envelope?.cipher === "age-v1") ageV1Envelopes += 1; diff --git a/tests/remote_sync_engine.test.mjs b/tests/remote_sync_engine.test.mjs index 1dfe079fd..aea2de908 100644 --- a/tests/remote_sync_engine.test.mjs +++ b/tests/remote_sync_engine.test.mjs @@ -4085,3 +4085,125 @@ test("runLoop: a non-retryable error that is NOT a refusal still ends the loop", eventCall: async () => {}, }), /config is unreadable/); }); + +test("pull bootstrap reports progress on stderr and leaves stdout as the result channel", async () => { + const teamId = "018f3f7e-0000-7000-8000-000000000001"; + const serverId = "018f3f7e-0000-7000-8000-000000000002"; + // Both streams are captured, not just the one under test. `cmd_pull` reads + // this process's stdout as the result -- result="$(... pull-bootstrap ...)" + // then greps it for pull_bootstrap_result -- so a progress line landing there + // is the regression this case exists to catch, and it is invisible unless + // stdout is measured too. + const out = []; + const err = []; + const realOut = process.stdout.write.bind(process.stdout); + const realErr = process.stderr.write.bind(process.stderr); + process.stdout.write = (chunk) => { out.push(String(chunk)); return true; }; + process.stderr.write = (chunk) => { err.push(String(chunk)); return true; }; + try { + await pullBootstrap({ + team: "clone", "team-id": teamId, + // The shape a hosted endpoint really has: the path IS the capability. + endpoint: "https://user:pa55word@sync.example.test:8443/t/agsy_SECRETCAP123?q=1#f", + }, { + publicSnapshotCall: async () => ({ + server_instance_id: serverId, team_id: teamId, team_name: "source", + min_available_seq: "0", + }), + requestPublicCall: async () => ({ + messages: [{ id: "01", seq: "1", envelope: { v: 1, cipher: "plain", blob: "x" } }], + next_after: "1", has_more: false, + }), + evaluateCall: async () => ({ + status: "importable", policy_revision: "0", local_security_revision: "0", + }), + driverCall: async () => [{ type: "sync_apply_result", transport_cursor: "1", corrupt_count: 0 }], + rosterDriverCall: async () => [], + eventCall: async () => {}, + }); + } finally { + process.stdout.write = realOut; + process.stderr.write = realErr; + } + + // stdout: exactly the result, still parseable as one JSON line. + const stdoutLines = out.join("").split("\n").filter((line) => line !== ""); + assert.equal(stdoutLines.length, 1); + assert.equal(JSON.parse(stdoutLines[0]).type, "pull_bootstrap_result"); + + // stderr: the operator can see it start, and can see it move. Both halves are + // named, because when this stops moving the line it stopped on says whether + // to look at the network or at the driver's child process. + const stderrText = err.join(""); + assert.match(stderrText, /agmsg: \[\d+s\] pulling clone from sync\.example\.test:8443 /); + // WHAT MUST NOT BE THERE, named one piece at a time. This is the line a person + // pastes into an issue when a pull is taking too long, so the capability in + // the path, the credential before the host, and the query and fragment beside + // them all have to be absent -- and asserting the host is present does not say + // that any of them are gone. + for (const secret of ["agsy_SECRETCAP123", "pa55word", "/t/", "q=1", "#f"]) { + assert.ok(!stderrText.includes(secret), `stderr must not carry ${secret}`); + } + assert.ok(!out.join("").includes("agsy_SECRETCAP123"), "stdout must not carry it either"); + assert.match(stderrText, /agmsg: \[\d+s\] fetching messages after /); + assert.match(stderrText, /agmsg: \[\d+s\] applying 1 messages/); +}); + +test("pull bootstrap prints a server cursor only when it is a canonical sequence", async () => { + // WHERE A MALFORMED CURSOR ACTUALLY COMES FROM. The first cursor is not ours: + // it is `teamSnapshot.min_available_seq`, and `publicSnapshot` checks only the + // team id and the server instance id -- the sequence is never validated, so a + // server's value reaches the first progress line exactly as it was sent. + // + // The second page cannot be the case this pins, even though it looks like the + // better one. A malformed `next_after` would have to survive the driver first, + // and the real sqlite driver refuses a non-numeric `sync_pull_cursor` + // (sqlite-sync.sh:891-897, `return 13`) before the loop comes round again. A + // fixture built there is testing a path only a stubbed driver allows. + // + // The line matters because it is the one people paste when a pull is slow: an + // escape sequence in a pasted log is a terminal doing what the server's + // operator told it. + const ESC = String.fromCharCode(27); + const evil = `${ESC}[2Jwiped`; + const teamId = "018f3f7e-0000-7000-8000-000000000001"; + + const run = async (minAvailableSeq) => { + const err = []; + const realErr = process.stderr.write.bind(process.stderr); + const realOut = process.stdout.write.bind(process.stdout); + process.stderr.write = (chunk) => { err.push(String(chunk)); return true; }; + process.stdout.write = () => true; + try { + await pullBootstrap({ + team: "clone", "team-id": teamId, endpoint: "https://sync.example.test/t/agsy_X", + }, { + publicSnapshotCall: async () => ({ + server_instance_id: "018f3f7e-0000-7000-8000-000000000002", + team_id: teamId, team_name: "source", min_available_seq: minAvailableSeq, + }), + requestPublicCall: async () => ({ messages: [], next_after: "9", has_more: false }), + evaluateCall: async () => ({ status: "importable" }), + driverCall: async () => [{ type: "sync_apply_result", transport_cursor: "9", corrupt_count: 0 }], + rosterDriverCall: async () => [], + eventCall: async () => {}, + }); + } finally { + process.stderr.write = realErr; + process.stdout.write = realOut; + } + return err.join(""); + }; + + const bad = await run(evil); + assert.match(bad, /fetching messages after an unreadable cursor /); + assert.ok(!bad.includes(ESC), "no escape byte reaches stderr"); + assert.ok(!bad.includes("wiped"), "and nothing that rode with it"); + + // The control on the replacement: a sequence prints as itself. Without this a + // guard that had decayed into printing the placeholder for everything would + // satisfy every assertion above -- redacting and erasing are not the same act. + const good = await run("41"); + assert.match(good, /fetching messages after 41 /); + assert.ok(!good.includes("an unreadable cursor"), "a real sequence is not replaced"); +}); diff --git a/tests/test_remote_sync.bats b/tests/test_remote_sync.bats index 408f48027..09fd8edf3 100644 --- a/tests/test_remote_sync.bats +++ b/tests/test_remote_sync.bats @@ -861,3 +861,77 @@ second line after a tab" [ "$(sqlite3 "$db" "SELECT count(*) FROM events WHERE body IN ('hidden in front','the visible one');" | tr -d '\r')" -eq 0 ] [ "$(sqlite3 "$db" "SELECT count(*) FROM sync_quarantine WHERE wire_id IN ('550e8400-e29b-41d4-a716-4466554400e1','550e8400-e29b-41d4-a716-4466554400e2');" | tr -d '\r')" -eq 0 ] } + +# Builds a pull page of N distinct importable messages, so the only thing that +# varies between two runs of the case below is how many ids the apply carries. +_sync_page_of() { + local n="$1" i seq id + i=0 + while [ "$i" -lt "$n" ]; do + seq=$((i + 1)) + id="$(printf '550e8400-e29b-41d4-a716-4466%08x' "$seq")" + jq -nc --arg id "$id" --arg seq "$seq" ' + {type:"sync_pull_message",server_seq:$seq,id:$id, + server_received_at:"2026-07-20T13:00:00.000000Z", + envelope:{v:1,cipher:"none",key_id:null,blob:( + {body:("m" + $seq),created_at:"2026-07-20T13:00:00.000000Z", + from_agent:"carol",to_agent:"bob"}|tojson|@base64)}, + status:"importable",policy_revision:"0",local_security_revision:"0", + projection:{body:("m" + $seq),created_at:"2026-07-20T13:00:00.000000Z", + from_agent:"carol",to_agent:"bob"}}' + i=$((i + 1)) + done + jq -nc --arg after "$n" '{type:"sync_pull_cursor",next_after:$after}' +} + +# Records the length of every sqlite3 command line, then runs the real one. +# Measuring the ARGUMENT LENGTH rather than waiting for an operating system to +# refuse it is what makes this case mean the same thing on every platform: the +# limit that broke #882 is Windows' 32,767 characters, and a test that only +# went red where the limit is small would be green on the machines that run it. +_sqlite_argv_recorder() { + local dir="$TEST_SKILL_DIR/argv-probe" real + real="$(command -v sqlite3)" + mkdir -p "$dir" + : > "$dir/lengths" + printf '%s\n' '#!/usr/bin/env bash' \ + 'joined="$*"' \ + "printf '%s\\n' \"\${#joined}\" >> $(printf '%q' "$dir/lengths")" \ + "exec $(printf '%q' "$real") \"\$@\"" > "$dir/sqlite3" + chmod +x "$dir/sqlite3" + printf '%s\n' "$dir" +} + +_longest_argv() { + sort -n "$TEST_SKILL_DIR/argv-probe/lengths" | tail -1 +} + +@test "sync contract: applying a pull page does not grow the command line (#882)" { + # A Windows machine could not pull a team past a few hundred messages: the + # apply put one wire id per message into the SQL that reads the outcomes back, + # twice, and handed it to sqlite3 as an ARGUMENT. Measured on Windows, sqlite3 + # took 827 uuids and refused 837; the command line caps at 32,767 characters. + # Nothing in the product chose that number, and the team that hit it was 2,079 + # messages. + local probe short long + probe="$(_sqlite_argv_recorder)" + + _sync_page_of 5 | PATH="$probe:$PATH" storage_sync_apply_pull demo "$SERVER_ID" "$TEAM_ID" 1 >/dev/null + short="$(_longest_argv)" + + : > "$TEST_SKILL_DIR/argv-probe/lengths" + _sync_page_of 120 | PATH="$probe:$PATH" storage_sync_apply_pull demo "$SERVER_ID" "$TEAM_ID" 1 >/dev/null + long="$(_longest_argv)" + + # The page really did get bigger -- without this the case would pass if both + # runs had silently applied nothing. + # The second page is a superset of the first, so the store holds 120 -- what + # this pins is that the larger apply really did import, which is what makes + # the length comparison above about anything. + [ "$(storage_history demo | jq -s 'length')" -eq 120 ] + + # Before the fix this difference was 115 messages x 78 characters. The bound + # is deliberately loose: what must not happen is growth PER MESSAGE, and a + # cursor or a count moving by a few characters is not that. + [ "$long" -lt "$((short + 200))" ] +}