From 2c61b3a352521f95a726c461db8246f4874559eb Mon Sep 17 00:00:00 2001 From: Todd Hebebrand Date: Sat, 8 Aug 2026 00:05:48 -0600 Subject: [PATCH 1/3] fix(backup): bound the command result against the SERVER's 1 MiB cap, not the IPC frame (#3001) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The #3001 residual, reproduced on v0.104.0: a 4,000-file backup completes on the endpoint, its terminal result never reaches the API, and the stale-backup reaper fails a job that succeeded. A 1,200-file run lands normally. Nothing is logged anywhere on either side. Root cause. The result is refused by `commandResultSchema`'s 1 MiB cap on the `result` field (apps/api/src/routes/agents/schemas.ts). The backup helper's stdout is the full BackupJob JSON including one `snapshot.files` entry per backed-up file (~522 B each), and the forwarder assigns that body to `result` rather than `stdout` — which has a 5 MB budget. 1048576/522 puts the cliff at ~2,008 files, exactly inside the observed 1,200-passes / 4,000-fails bracket. The silence had three independent causes, all fixed here: - the helper's tiered degradation bounded against the 16 MiB IPC frame — the next hop, not the binding limit — so it never fired and never logged; - the server logged the rejection as a generic invalid-message with no commandId, no size and no type, while both backup-specific log lines live downstream of the failed parse and never ran; - the server's error frame carries no `id`, so the agent's readPump discarded it under the "not a command" skip. Changes: - agent/internal/wire: new leaf package mirroring the server's cap, pinned to the TypeScript declaration by a test on each side. - helper: bound Stdout against the server budget as well as the IPC frame, so tier 2 drops the per-file index at the limit that actually binds. A 100k-file backup now reports completion with its snapshot identity intact. - WS client: generic backstop replacing an over-cap `result` body with a marker so ANY command type keeps its terminal status; and readPump now logs server rejections instead of dropping them. - server: a rejected command_result logs at error with commandId, frame size, measured result size and the limit, and echoes commandId in the reply. - the degradation log line names the limit that actually fired instead of always claiming the IPC frame (it reported limitBytes=16777216 for a 10 KB payload truncated by the 8 KiB stderr cap). Co-Authored-By: Claude Opus 4.8 (1M context) --- .../partial_status_bounds_test.go | 2 +- agent/cmd/breeze-backup/result_bounds.go | 144 +++++++++++-- .../result_bounds_server_cap_test.go | 193 ++++++++++++++++++ agent/cmd/breeze-backup/result_bounds_test.go | 26 +-- .../result_warning_preservation_test.go | 2 +- agent/internal/websocket/client.go | 122 ++++++++++- .../websocket/result_server_cap_test.go | 189 +++++++++++++++++ agent/internal/wire/limits.go | 59 ++++++ agent/internal/wire/limits_test.go | 75 +++++++ apps/api/src/routes/agentWs.ts | 42 +++- .../agents/schemas.commandResult.test.ts | 137 +++++++++++++ apps/api/src/routes/agents/schemas.ts | 57 +++++- 12 files changed, 1007 insertions(+), 41 deletions(-) create mode 100644 agent/cmd/breeze-backup/result_bounds_server_cap_test.go create mode 100644 agent/internal/websocket/result_server_cap_test.go create mode 100644 agent/internal/wire/limits.go create mode 100644 agent/internal/wire/limits_test.go create mode 100644 apps/api/src/routes/agents/schemas.commandResult.test.ts diff --git a/agent/cmd/breeze-backup/partial_status_bounds_test.go b/agent/cmd/breeze-backup/partial_status_bounds_test.go index 0680f61fc..0ead1b9d4 100644 --- a/agent/cmd/breeze-backup/partial_status_bounds_test.go +++ b/agent/cmd/breeze-backup/partial_status_bounds_test.go @@ -25,7 +25,7 @@ import ( // tier that fired. func fitPartial(t *testing.T, in backupipc.BackupCommandResult) (status string, degraded string, fitted backupipc.BackupCommandResult) { t.Helper() - fitted, degraded = fitBackupResultToIPC(in) + fitted, degraded = fitBackupResultForDelivery(in) if got := payloadSize(t, fitted); got > resultPayloadBudget { t.Fatalf("fitted payload is %d bytes, over the %d budget", got, resultPayloadBudget) } diff --git a/agent/cmd/breeze-backup/result_bounds.go b/agent/cmd/breeze-backup/result_bounds.go index e79c28b53..b6f9c765b 100644 --- a/agent/cmd/breeze-backup/result_bounds.go +++ b/agent/cmd/breeze-backup/result_bounds.go @@ -9,6 +9,7 @@ import ( "github.com/breeze-rmm/agent/internal/backupipc" "github.com/breeze-rmm/agent/internal/ipc" "github.com/breeze-rmm/agent/internal/logging" + "github.com/breeze-rmm/agent/internal/wire" ) // Issue #3001: a backup run's terminal result is unbounded, but the IPC frame @@ -24,6 +25,20 @@ import ( // // Deliberately NOT fixed by raising ipc.MaxMessageSize: the payload has to be // bounded regardless, and a bigger cap only moves the cliff. +// +// #3001 RESIDUAL (v0.104.0). The above fixed the loud failure and missed a +// quiet one, for a reason worth stating plainly: this file bounded against the +// NEXT HOP rather than the DESTINATION. The IPC frame is merely the first of +// four limits the result passes; the tightest is the server's 1 MiB cap on the +// command_result `result` field, 16x below the budget used here. So a +// 4,000-file run (~2 MB of snapshot index) cleared every check in this file, +// was written to the socket successfully, and was refused on arrival — with no +// error logged on either side, because the send had genuinely succeeded and the +// server's rejection was an id-less frame the agent discarded. The job sat +// `running` until the reaper failed a backup that had completed. +// +// The rule the fix encodes: bound against the tightest limit anywhere on the +// path (see serverResultBudget), never against the one nearest to hand. const ( // resultEnvelopeHeadroom reserves room for the ipc.Envelope fields wrapped @@ -36,6 +51,25 @@ const ( // hand to conn.Send. resultPayloadBudget = ipc.MaxMessageSize - resultEnvelopeHeadroom + // serverResultBudget is the largest Stdout body the SERVER will accept, and + // it — not resultPayloadBudget — is what actually binds. + // + // #3001 residual: bounding against the next hop is not the same as bounding + // against the destination. This file shipped bounding only against the + // 16 MiB IPC frame, but Stdout does not stop at the agent: the forwarder + // (internal/heartbeat, case TypeBackupResult) parses it and assigns it to + // the `result` field of the WS command_result, where the server caps it at + // wire.MaxCommandResultBytes — 16x tighter. So every tier below was dead + // code for the failure that mattered: a 4,000-file run built a ~2 MB + // snapshot index, sailed through a 15.9 MiB budget, and was refused by the + // server. Nothing logged on either side and the job was reaped as stalled. + // + // Checked against Stdout alone rather than the whole marshalled result + // because that is the field the cap applies to server-side; Stderr rides + // the separate `error` field (capped at 10,000 there, and already held to + // maxResultTextBytes here). + serverResultBudget = wire.CommandResultBudget + // maxResultTextBytes caps free-text fields (warning, stderr) that are built // by joining per-file errors. summarizeUploadFailures already caps the // number of detail entries it renders, but a hard-failed run routes @@ -113,8 +147,9 @@ var snapshotIdentityKeys = []string{"id", "timestamp", "size", "formatVersion", // Anything under it is cheaper to keep than to reason about. const bulkFieldThreshold = 4 * 1024 -// fitBackupResultToIPC returns result bounded so its marshalled payload fits -// resultPayloadBudget, degrading in tiers and stopping at the first that fits: +// fitBackupResultForDelivery returns result bounded so it fits EVERY limit on +// the path to the server — serverResultBudget as well as resultPayloadBudget — +// degrading in tiers and stopping at the first that fits: // // 1. always: truncate the free-text stderr field (cheap, no JSON parse); // 2. empty the per-file snapshot index, cap the warning text, and drop any @@ -138,9 +173,31 @@ const bulkFieldThreshold = 4 * 1024 // The second return value describes what was dropped, and is "" when the result // was already within budget. It is non-empty exactly when the caller should log // loudly — a degraded result is a real (if survivable) loss of detail. -func fitBackupResultToIPC(result backupipc.BackupCommandResult) (backupipc.BackupCommandResult, string) { +func fitBackupResultForDelivery(result backupipc.BackupCommandResult) (backupipc.BackupCommandResult, string) { + fitted, notes, _, _ := fitBackupResult(result) + return fitted, notes +} + +// fitBackupResult is fitBackupResultForDelivery plus the attribution the log line +// needs: the name and value of the limit that forced the degradation. +// +// #3001: the caller used to log ipc.MaxMessageSize unconditionally, so a run +// degraded ONLY by the 8 KiB stderr cap reported "exceeded the IPC limit … +// sentBytes=10195 limitBytes=16777216" — a 10 KB payload described as +// overflowing a 16 MiB frame. Naming a limit 2,000x larger than the one that +// fired is worse than saying nothing: it asserts, wrongly, that the frame was +// the problem. The trigger is therefore tracked as the tiers run rather than +// assumed at the end. +func fitBackupResult(result backupipc.BackupCommandResult) ( + fitted backupipc.BackupCommandResult, notesText, limitName string, limitBytes int, +) { var notes []string + // Captured BEFORE any tier runs: once the tiers have shrunk the payload the + // binding limit is no longer observable, and reporting a limit derived from + // the already-degraded result is how the wrong one gets named. + limitName, limitBytes = exceededLimit(result) + // Tier 1 — always applied, and cheap (no JSON parse). Bounding the failure // detail BEFORE marshalling is the primary fix; the tiers below are the // defensive net behind it. Stderr is the unbounded one on this path: a hard @@ -152,7 +209,7 @@ func fitBackupResultToIPC(result backupipc.BackupCommandResult) (backupipc.Backu notes = append(notes, fmt.Sprintf("stderr truncated (%d bytes dropped)", dropped)) } if fits(result) { - return result, joinNotes(notes) + return finishBounding(result, notes, limitName, limitBytes) } // Only reached when the result is actually oversize, so the ordinary path @@ -160,7 +217,7 @@ func fitBackupResultToIPC(result backupipc.BackupCommandResult) (backupipc.Backu obj, isObject := decodeStdoutObject(result.Stdout) if !isObject { notes = append(notes, "result body could not be summarised and was replaced with an oversize failure") - return oversizeFailureResult(result), joinNotes(notes) + return finishBounding(oversizeFailureResult(result), notes, limitName, limitBytes) } // Tier 2 — cap the warning text, empty the per-file snapshot index, and @@ -184,7 +241,7 @@ func fitBackupResultToIPC(result backupipc.BackupCommandResult) (backupipc.Backu } result.Stdout = encodeStdoutObject(obj, result.Stdout) if fits(result) { - return result, joinNotes(notes) + return finishBounding(result, notes, limitName, limitBytes) } // Tier 3 — scalar fields plus the snapshot identity only. @@ -192,7 +249,7 @@ func fitBackupResultToIPC(result backupipc.BackupCommandResult) (backupipc.Backu result.Stdout = reduced notes = append(notes, "result reduced to summary scalars only") if fits(result) { - return result, joinNotes(notes) + return finishBounding(result, notes, limitName, limitBytes) } } @@ -202,7 +259,24 @@ func fitBackupResultToIPC(result backupipc.BackupCommandResult) (backupipc.Backu result.Stdout = lastResortStdout(result.Stdout) result.Stderr, _ = truncateText(result.Stderr, maxLastResortFieldBytes) result.CommandID, _ = truncateText(result.CommandID, maxLastResortFieldBytes) - return result, joinNotes(notes) + return finishBounding(result, notes, limitName, limitBytes) +} + +// finishBounding assembles fitBackupResult's return, attributing the +// degradation to the free-text cap when no delivery limit was breached. +// +// That branch is the whole point of the attribution: tier 1 runs on EVERY +// result and fires whenever stderr is over 8 KiB, including on results that +// were always going to fit the wire comfortably. Those are the runs the old log +// line mislabelled as frame overflows. +func finishBounding( + result backupipc.BackupCommandResult, notes []string, limitName string, limitBytes int, +) (backupipc.BackupCommandResult, string, string, int) { + notesText := joinNotes(notes) + if notesText != "" && limitName == "" { + limitName, limitBytes = limitResultText, maxResultTextBytes + } + return result, notesText, limitName, limitBytes } // sendBackupResult sends a terminal backup_result envelope, bounding the @@ -214,18 +288,22 @@ func fitBackupResultToIPC(result backupipc.BackupCommandResult) (backupipc.Backu // log-shipping threshold and therefore on the server. func sendBackupResult(conn *ipc.Conn, envelopeID string, result backupipc.BackupCommandResult) error { log := logging.L("backup") - fitted, degraded := fitBackupResultToIPC(result) + fitted, degraded, limitName, limitBytes := fitBackupResult(result) if degraded != "" { // Sizes are reported from the raw fields rather than a full marshal of // the original: re-marshalling tens of megabytes just to populate a log // field is not worth it at the tail of a backup run. - log.Error("backup result payload exceeded the IPC limit and was degraded to fit", + // + // The message no longer claims which hop overflowed — limitName says + // that, and it is frequently NOT the IPC frame. + log.Error("backup result payload was degraded to fit a delivery limit", "commandId", result.CommandID, "degraded", degraded, + "limitName", limitName, + "limitBytes", limitBytes, "originalStdoutBytes", len(result.Stdout), "originalStderrBytes", len(result.Stderr), "sentBytes", marshalledSize(fitted), - "limitBytes", ipc.MaxMessageSize, ) } if err := conn.SendTyped(envelopeID, backupipc.TypeBackupResult, fitted); err != nil { @@ -242,13 +320,33 @@ func sendBackupResult(conn *ipc.Conn, envelopeID string, result backupipc.Backup // --- helpers --- -// fits reports whether result's marshalled payload is within budget. The raw -// text fields are checked first as a cheap lower bound — JSON string encoding -// never shrinks its input — so an oversize result is rejected without -// marshalling tens of megabytes on the endpoint. +// fits reports whether result is within every delivery limit on the path to +// the server. func fits(result backupipc.BackupCommandResult) bool { + name, _ := exceededLimit(result) + return name == "" +} + +// exceededLimit names the first delivery limit result's payload breaches, and +// that limit's value; it returns ("", 0) when the result is deliverable. +// +// Both limits are checked, tightest first, because "which limit fired" is not a +// detail — it is the whole diagnosis. A run degraded by the server's `result` +// cap and a run degraded by the IPC frame are different bugs on different hops, +// and the log line that conflated them (see sendBackupResult) sent #3001's +// investigation to the wrong layer twice. +// +// The raw text fields are measured before marshalling as a cheap lower bound — +// JSON string encoding never shrinks its input — so an oversize result is +// rejected without marshalling tens of megabytes on the endpoint. +func exceededLimit(result backupipc.BackupCommandResult) (string, int) { + // Tightest first. Stdout alone, because Stdout is what becomes the server's + // `result` field; see serverResultBudget. + if len(result.Stdout) > serverResultBudget { + return limitServerResult, wire.MaxCommandResultBytes + } if len(result.Stdout)+len(result.Stderr)+len(result.CommandID) > resultPayloadBudget { - return false + return limitIPCFrame, ipc.MaxMessageSize } size := marshalledSize(result) // A result that cannot be marshalled at all is not "fitting" — treating @@ -256,12 +354,20 @@ func fits(result backupipc.BackupCommandResult) bool { // payload from a function whose contract is that the send will succeed. // (BackupCommandResult is only strings/bool/int64, so this is unreachable // today; it is guarded so it stays unreachable if the type grows a field.) - if size < 0 { - return false + if size < 0 || size > resultPayloadBudget { + return limitIPCFrame, ipc.MaxMessageSize } - return size <= resultPayloadBudget + return "", 0 } +// Limit names used in the degradation log line. They are the identifiers an +// operator can grep for in this repo, not prose. +const ( + limitServerResult = "server command_result `result` cap (wire.MaxCommandResultBytes)" + limitIPCFrame = "agent IPC frame (ipc.MaxMessageSize)" + limitResultText = "free-text field cap (maxResultTextBytes)" +) + // marshalledSize returns the marshalled byte length of result, or -1 when it // cannot be marshalled at all (which conn.Send would reject anyway). func marshalledSize(result backupipc.BackupCommandResult) int { diff --git a/agent/cmd/breeze-backup/result_bounds_server_cap_test.go b/agent/cmd/breeze-backup/result_bounds_server_cap_test.go new file mode 100644 index 000000000..3960d92b4 --- /dev/null +++ b/agent/cmd/breeze-backup/result_bounds_server_cap_test.go @@ -0,0 +1,193 @@ +package main + +import ( + "encoding/json" + "strings" + "testing" + + "github.com/breeze-rmm/agent/internal/backupipc" + "github.com/breeze-rmm/agent/internal/ipc" + "github.com/breeze-rmm/agent/internal/wire" +) + +// qaResidualFileCount is the file count from the #3001 residual reproduction on +// v0.104.0: two 4,000-file / 200 MB runs whose terminal result never arrived, +// against a 1,200-file run that landed normally. The loss threshold sat between +// them, which is 1_048_576 / ~522 B-per-entry ≈ 2,008 files. +const ( + qaResidualFileCount = 4000 + qaPassingFileCount = 1200 +) + +// TestFourThousandFileRunIsDegradedForTheServerCap is the residual #3001 +// regression, and the one that would have caught it. +// +// The previous bounding stopped at the 15.9 MiB IPC budget, so a ~2 MB result +// "fitted" and was sent verbatim — then refused by the server's 1 MiB `result` +// cap with no log on either side, and the job was reaped as stalled 15 minutes +// after a backup that had SUCCEEDED. The fixture is deliberately the size that +// passed the old check and failed the real one. +func TestFourThousandFileRunIsDegradedForTheServerCap(t *testing.T) { + result := mustRunResult(t, buildLargeRunJob(qaResidualFileCount, 3)) + + if len(result.Stdout) > resultPayloadBudget { + t.Fatalf("fixture stdout is %d bytes, over the IPC budget %d — this test must exercise a "+ + "payload the OLD (IPC-only) bounding considered acceptable", len(result.Stdout), resultPayloadBudget) + } + if len(result.Stdout) <= wire.MaxCommandResultBytes { + t.Fatalf("fixture stdout is only %d bytes, under the server cap %d — the test would prove nothing", + len(result.Stdout), wire.MaxCommandResultBytes) + } + + fitted, notes, limitName, limitBytes := fitBackupResult(result) + + if notes == "" { + t.Fatal("a result over the server's `result` cap was passed through undegraded — this is #3001") + } + if len(fitted.Stdout) > serverResultBudget { + t.Fatalf("degraded stdout is %d bytes, still over the server budget %d", + len(fitted.Stdout), serverResultBudget) + } + if limitName != limitServerResult { + t.Fatalf("degradation attributed to %q, want the server result cap", limitName) + } + if limitBytes != wire.MaxCommandResultBytes { + t.Fatalf("reported limitBytes = %d, want %d", limitBytes, wire.MaxCommandResultBytes) + } + if !strings.Contains(notes, "snapshot file index dropped") { + t.Fatalf("expected the per-file index to be the thing dropped, got notes %q", notes) + } + + assertTerminalStatusSurvives(t, fitted, result.CommandID) +} + +// TestTwelveHundredFileRunIsSentIntact is the other half of the QA +// reproduction: the run that WORKED must keep working. A fix that degrades +// every backup would "pass" the test above while destroying restore browsing +// for the endpoints that never had a problem. +func TestTwelveHundredFileRunIsSentIntact(t *testing.T) { + result := mustRunResult(t, buildLargeRunJob(qaPassingFileCount, 0)) + + fitted, notes, limitName, _ := fitBackupResult(result) + + if notes != "" { + t.Fatalf("a 1,200-file run was degraded (%q); it fits the server cap and must be sent intact", notes) + } + if limitName != "" { + t.Fatalf("no limit should have been reported for an in-budget result, got %q", limitName) + } + if fitted.Stdout != result.Stdout { + t.Fatal("stdout was modified for an in-budget result") + } + + var job map[string]any + if err := json.Unmarshal([]byte(fitted.Stdout), &job); err != nil { + t.Fatalf("unmarshal fitted stdout: %v", err) + } + snap, _ := job["snapshot"].(map[string]any) + files, _ := snap["files"].([]any) + if len(files) != qaPassingFileCount { + t.Fatalf("file index has %d entries, want the full %d — restore browsing must survive here", + len(files), qaPassingFileCount) + } +} + +// TestHundredThousandFileRunStillReportsCompletion is fix requirement 1 stated +// as a test: "a clean backup of 100k+ files must be able to report completion". +func TestHundredThousandFileRunStillReportsCompletion(t *testing.T) { + result := mustRunResult(t, buildLargeRunJob(100000, 0)) + + fitted, notes, _, _ := fitBackupResult(result) + + if notes == "" { + t.Fatal("a 100k-file result was not degraded at all") + } + if len(fitted.Stdout) > serverResultBudget { + t.Fatalf("degraded stdout is %d bytes, over the server budget %d — the server would refuse it "+ + "and the job would be reaped as stalled", len(fitted.Stdout), serverResultBudget) + } + assertTerminalStatusSurvives(t, fitted, result.CommandID) + + // The snapshot IDENTITY is what makes the run a usable restore point even + // with no browsable index, so it must outlive the degradation. + var job map[string]any + if err := json.Unmarshal([]byte(fitted.Stdout), &job); err != nil { + t.Fatalf("unmarshal fitted stdout: %v", err) + } + snap, ok := job["snapshot"].(map[string]any) + if !ok { + t.Fatalf("snapshot object did not survive degradation: %v", job) + } + if snap["id"] != "snapshot-20260801T125517Z-5edfcd7e" { + t.Fatalf("snapshot id did not survive degradation: %v", snap["id"]) + } +} + +// TestStderrOnlyDegradationNamesTheTextCap is fix requirement 3. +// +// The reported symptom was a log line reading "exceeded the IPC limit … +// sentBytes=10195 limitBytes=16777216": a 10 KB payload described as +// overflowing a 16 MiB frame, because the message and the limit were both +// hardcoded. Here nothing but the 8 KiB free-text cap fires, and that is what +// must be named. +func TestStderrOnlyDegradationNamesTheTextCap(t *testing.T) { + result := backupipc.BackupCommandResult{ + CommandID: "822e0c7f-7e35-43e6-b0fc-0912a5c0d221", + Success: false, + Stdout: `{"id":"job-1","status":"failed"}`, + Stderr: strings.Repeat("access is denied: C:\\Users\\jdoe\\ntuser.dat; ", 400), + } + if len(result.Stderr) <= maxResultTextBytes { + t.Fatalf("fixture stderr is %d bytes, under the %d text cap — nothing would fire", + len(result.Stderr), maxResultTextBytes) + } + + fitted, notes, limitName, limitBytes := fitBackupResult(result) + + if notes == "" { + t.Fatal("oversize stderr was not truncated") + } + if limitName != limitResultText { + t.Fatalf("attributed the degradation to %q; the free-text cap is what fired", limitName) + } + if limitBytes != maxResultTextBytes { + t.Fatalf("reported limitBytes = %d, want the text cap %d", limitBytes, maxResultTextBytes) + } + if limitBytes == ipc.MaxMessageSize { + t.Fatal("still reporting the IPC frame size for a degradation the IPC frame did not cause") + } + if marshalledSize(fitted) > serverResultBudget { + t.Fatalf("fitted result is %d bytes, over budget", marshalledSize(fitted)) + } +} + +// TestDeliveryWrapperMatchesAttributedForm keeps the two-return +// wrapper honest, so the existing suite that calls it keeps testing the code +// the sender actually runs. +func TestDeliveryWrapperMatchesAttributedForm(t *testing.T) { + result := mustRunResult(t, buildLargeRunJob(qaResidualFileCount, 3)) + + wrappedResult, wrappedNotes := fitBackupResultForDelivery(result) + fullResult, fullNotes, _, _ := fitBackupResult(result) + + if wrappedNotes != fullNotes { + t.Fatalf("wrapper notes %q != %q", wrappedNotes, fullNotes) + } + if wrappedResult.Stdout != fullResult.Stdout || wrappedResult.Stderr != fullResult.Stderr { + t.Fatal("wrapper returned a different result than fitBackupResult") + } +} + +// assertTerminalStatusSurvives checks the invariant every tier exists to +// protect: whatever else is dropped, the server must still learn that the +// command finished and how. +func assertTerminalStatusSurvives(t *testing.T, fitted backupipc.BackupCommandResult, commandID string) { + t.Helper() + if !fitted.Success { + t.Fatal("Success flag was lost during degradation; a succeeded backup would report as failed") + } + if fitted.CommandID != commandID { + t.Fatalf("CommandID = %q, want %q — the server cannot attribute a result with no command id", + fitted.CommandID, commandID) + } +} diff --git a/agent/cmd/breeze-backup/result_bounds_test.go b/agent/cmd/breeze-backup/result_bounds_test.go index 7b4fc000c..da6399745 100644 --- a/agent/cmd/breeze-backup/result_bounds_test.go +++ b/agent/cmd/breeze-backup/result_bounds_test.go @@ -75,7 +75,7 @@ func buildLargeRunJob(files, failures int) *backup.BackupJob { // Building and marshalling a 60k-entry manifest costs ~5s under -race, and // several tests need the identical input, so the marshalled stdout is built // once per package run. The fixtures are treated strictly read-only (Go strings -// are immutable and fitBackupResultToIPC takes its argument by value). +// are immutable and fitBackupResultForDelivery takes its argument by value). var ( oversizeRunOnce sync.Once oversizeRunStdout string @@ -155,10 +155,10 @@ func TestUnboundedRunResultExceedsIPCLimit(t *testing.T) { // TestFitBackupResultBoundsLargeRun is the core contract: whatever the run // produced, the result handed to conn.Send fits the frame. func TestFitBackupResultBoundsLargeRun(t *testing.T) { - fitted, degraded := fitBackupResultToIPC(oversizeRunResult(t)) + fitted, degraded := fitBackupResultForDelivery(oversizeRunResult(t)) if degraded == "" { - t.Fatal("expected fitBackupResultToIPC to report that it degraded the payload") + t.Fatal("expected fitBackupResultForDelivery to report that it degraded the payload") } if got := payloadSize(t, fitted); got > resultPayloadBudget { t.Fatalf("fitted payload is %d bytes, over the %d budget", got, resultPayloadBudget) @@ -171,7 +171,7 @@ func TestFitBackupResultBoundsLargeRun(t *testing.T) { // truncation. Losing error detail is acceptable; losing the fact that the // backup succeeded is not. func TestFitBackupResultPreservesTerminalStatus(t *testing.T) { - fitted, _ := fitBackupResultToIPC(oversizeRunResult(t)) + fitted, _ := fitBackupResultForDelivery(oversizeRunResult(t)) if !fitted.Success { t.Error("expected Success to survive truncation") @@ -230,7 +230,7 @@ func TestFitBackupResultPreservesTerminalStatus(t *testing.T) { // turns into the browsable restore file list. func TestFitBackupResultKeepsFileIndexWhenItFits(t *testing.T) { job := buildLargeRunJob(500, 0) - fitted, degraded := fitBackupResultToIPC(mustRunResult(t, job)) + fitted, degraded := fitBackupResultForDelivery(mustRunResult(t, job)) if degraded != "" { t.Errorf("expected no degradation for a small run, got %q", degraded) @@ -257,7 +257,7 @@ func TestFitBackupResultBoundsOversizeStderr(t *testing.T) { huge := strings.Repeat("open C:\\Users\\jdoe\\file.dat: access is denied; ", 900000) result := backupipc.BackupCommandResult{CommandID: "cmd-1", Success: false, Stderr: huge} - fitted, degraded := fitBackupResultToIPC(result) + fitted, degraded := fitBackupResultForDelivery(result) if degraded == "" { t.Error("expected an oversize stderr to be reported as degraded") } @@ -324,7 +324,7 @@ func TestFitBackupResultAlwaysFits(t *testing.T) { } for name, tc := range cases { t.Run(name, func(t *testing.T) { - fitted, _ := fitBackupResultToIPC(tc.in) + fitted, _ := fitBackupResultForDelivery(tc.in) if got := payloadSize(t, fitted); got > resultPayloadBudget { t.Fatalf("fitted payload is %d bytes, over the %d budget", got, resultPayloadBudget) } @@ -419,7 +419,7 @@ func TestFitBackupResultPreservesRestoreScalars(t *testing.T) { } in := backupipc.BackupCommandResult{CommandID: "restore-1", Success: true, Stdout: string(data)} - fitted, degraded := fitBackupResultToIPC(in) + fitted, degraded := fitBackupResultForDelivery(in) if degraded == "" { t.Fatal("expected an oversize restore result to be reported as degraded") } @@ -469,7 +469,7 @@ func TestFitBackupResultRejectsUnsummarisableBody(t *testing.T) { Stdout: "[" + strings.Repeat(entry, 400000) + `{"id":"tail"}]`, } - fitted, degraded := fitBackupResultToIPC(in) + fitted, degraded := fitBackupResultForDelivery(in) if degraded == "" { t.Fatal("expected an oversize list result to be reported as degraded") } @@ -493,7 +493,7 @@ func TestFitBackupResultRejectsUnsummarisableBody(t *testing.T) { // would leave a previous delivery's backup_snapshot_files rows in place while // hasIndexedFiles flipped to false — two states that then disagree. func TestEmptySnapshotFilesKeepsTheFilesKey(t *testing.T) { - fitted, _ := fitBackupResultToIPC(oversizeRunResult(t)) + fitted, _ := fitBackupResultForDelivery(oversizeRunResult(t)) var out map[string]json.RawMessage if err := json.Unmarshal([]byte(fitted.Stdout), &out); err != nil { @@ -528,7 +528,7 @@ func TestFitBackupResultDropsOversizeSystemStateManifest(t *testing.T) { `"snapshot":{"id":"snapshot-1","size":345},"systemStateManifest":` + manifest + `}` in := backupipc.BackupCommandResult{CommandID: "sysimage-1", Success: true, Stdout: stdout} - fitted, degraded := fitBackupResultToIPC(in) + fitted, degraded := fitBackupResultForDelivery(in) if got := payloadSize(t, fitted); got > resultPayloadBudget { t.Fatalf("fitted payload is %d bytes, over the %d budget", got, resultPayloadBudget) } @@ -579,7 +579,7 @@ func TestFitBackupResultReducesToScalars(t *testing.T) { t.Fatalf("fixture must be oversize, got %d bytes", len(in.Stdout)) } - fitted, degraded := fitBackupResultToIPC(in) + fitted, degraded := fitBackupResultForDelivery(in) if got := payloadSize(t, fitted); got > resultPayloadBudget { t.Fatalf("fitted payload is %d bytes, over the %d budget", got, resultPayloadBudget) } @@ -641,7 +641,7 @@ func TestFitBackupResultLastResortRecoversStatus(t *testing.T) { b.WriteString("}") in := backupipc.BackupCommandResult{CommandID: "cmd-lastresort", Success: true, Stdout: b.String()} - fitted, degraded := fitBackupResultToIPC(in) + fitted, degraded := fitBackupResultForDelivery(in) if got := payloadSize(t, fitted); got > resultPayloadBudget { t.Fatalf("fitted payload is %d bytes, over the %d budget", got, resultPayloadBudget) } diff --git a/agent/cmd/breeze-backup/result_warning_preservation_test.go b/agent/cmd/breeze-backup/result_warning_preservation_test.go index 05fafbe2f..6ed9699bf 100644 --- a/agent/cmd/breeze-backup/result_warning_preservation_test.go +++ b/agent/cmd/breeze-backup/result_warning_preservation_test.go @@ -50,7 +50,7 @@ func atCapWarning() string { // tiers that fired. func fittedWarning(t *testing.T, in backupCommandResultFixture) (warning string, degraded string) { t.Helper() - fitted, degraded := fitBackupResultToIPC(in.result()) + fitted, degraded := fitBackupResultForDelivery(in.result()) if got := payloadSize(t, fitted); got > resultPayloadBudget { t.Fatalf("fitted payload is %d bytes, over the %d budget", got, resultPayloadBudget) } diff --git a/agent/internal/websocket/client.go b/agent/internal/websocket/client.go index b3400f8da..bc4ee17ae 100644 --- a/agent/internal/websocket/client.go +++ b/agent/internal/websocket/client.go @@ -17,6 +17,7 @@ import ( "github.com/breeze-rmm/agent/internal/netcache" "github.com/breeze-rmm/agent/internal/observability" "github.com/breeze-rmm/agent/internal/secmem" + "github.com/breeze-rmm/agent/internal/wire" ) var log = logging.L("websocket") @@ -441,10 +442,17 @@ func (c *Client) readPump() { continue } - // Skip non-command messages (ack, heartbeat_ack, error, etc.) + // A server rejection of something this agent sent. Handled BEFORE the + // id-less skip below, which used to swallow it (#3001). + if msg.Type == "error" { + logServerErrorFrame(message) + continue + } + + // Skip non-command messages (ack, heartbeat_ack, etc.) // Commands have both an ID and a type like "run_script", "list_processes", etc. if msg.ID == "" { - // Server acknowledgments, errors, etc. - not commands + // Server acknowledgments and other notices - not commands continue } @@ -742,6 +750,18 @@ func (c *Client) SendResult(result CommandResult) error { if err != nil { return fmt.Errorf("failed to marshal result: %w", err) } + if len(data) > wire.MaxCommandResultBytes { + // Only pay for the precise per-field measurement once the whole frame + // is over the cap — the `result` field is a subset of it, so a frame + // under the cap cannot contain an oversize field. + if bounded, ok := boundResultFieldForServer(result); ok { + result = bounded + data, err = json.Marshal(result) + if err != nil { + return fmt.Errorf("failed to marshal bounded result: %w", err) + } + } + } select { case c.resultChan <- outboundResult{data: data, result: result}: @@ -753,6 +773,104 @@ func (c *Client) SendResult(result CommandResult) error { } } +// resultOmittedMarker is the body substituted for an oversize `result`. The +// key is deliberately distinctive so it can be grepped for in stored command +// results and correlated with the agent-side error log. +const resultOmittedMarker = "_breezeResultOmitted" + +// boundResultFieldForServer replaces an oversize `result` body with a small +// marker describing what was dropped, returning the bounded result and whether +// any change was made. +// +// This is the LAST line of defence for #3001, and it is deliberately generic +// rather than backup-specific. The server rejects a command_result whose +// `result` field exceeds wire.MaxCommandResultBytes, and it rejects the WHOLE +// message when it does — status, exit code and error text included. So an +// oversize body did not merely lose its detail: it lost the fact that the +// command had finished at all, leaving the server to reap a job that had +// succeeded. Every command type shares that exposure (software inventory, +// patch scans, filesystem analysis are all `result` bodies that scale with the +// endpoint), not just backups. +// +// Trading the body for a marker is therefore always the right trade: losing +// the detail is bad, losing the terminal status is far worse. +// +// Producers should still bound their own payloads intelligently — the backup +// helper's tiers keep the snapshot identity and counters, which this cannot — +// so reaching here at all means a producer's own bounding was missing or wrong, +// and it logs at error accordingly. +func boundResultFieldForServer(result CommandResult) (CommandResult, bool) { + if result.Result == nil { + return result, false + } + encoded, err := json.Marshal(result.Result) + if err != nil { + log.Error("command result body cannot be marshalled; sending the terminal status without it", + "commandId", result.CommandID, + "status", result.Status, + "error", err.Error(), + ) + result.Result = map[string]any{ + resultOmittedMarker: true, + "reason": "result body could not be encoded", + } + return result, true + } + if len(encoded) <= wire.MaxCommandResultBytes { + return result, false + } + + log.Error("command result body exceeds the server's result cap and was dropped to preserve the terminal status", + "commandId", result.CommandID, + "status", result.Status, + "resultBytes", len(encoded), + "limitBytes", wire.MaxCommandResultBytes, + ) + result.Result = map[string]any{ + resultOmittedMarker: true, + "reason": "result body exceeded the server's command_result size limit", + "originalBytes": len(encoded), + "limitBytes": wire.MaxCommandResultBytes, + } + return result, true +} + +// logServerErrorFrame surfaces a server-side rejection of something this agent +// sent. +// +// Before #3001 these frames were discarded by the `msg.ID == ""` skip below, +// which is why a rejected terminal backup result produced NO agent-side trace +// whatsoever — the write succeeded, so every send path reported success, and +// the server's explanation was thrown away on arrival. An operator comparing +// agent and server logs saw a result that left the endpoint and never landed, +// with nothing anywhere naming a cause. +func logServerErrorFrame(raw []byte) { + var frame struct { + Code string `json:"code"` + Message string `json:"message"` + MessageType string `json:"messageType"` + CommandID string `json:"commandId"` + Details json.RawMessage `json:"details"` + } + if err := json.Unmarshal(raw, &frame); err != nil { + log.Error("server rejected a message and the error frame could not be parsed", + "error", err.Error()) + return + } + // Bounded so a verbose `details` array cannot flood the agent log. + details := string(frame.Details) + if len(details) > 2048 { + details = details[:2048] + "…(truncated)" + } + log.Error("server rejected a message sent by this agent", + "code", frame.Code, + "serverMessage", frame.Message, + "rejectedType", frame.MessageType, + "commandId", frame.CommandID, + "details", details, + ) +} + // handleResultWriteFailure hands a command result that writePump could not // deliver back to the outbox owner (if one is registered) so it can be // re-persisted for redelivery on the next reconnect. Safe to call with no diff --git a/agent/internal/websocket/result_server_cap_test.go b/agent/internal/websocket/result_server_cap_test.go new file mode 100644 index 000000000..5c3758c03 --- /dev/null +++ b/agent/internal/websocket/result_server_cap_test.go @@ -0,0 +1,189 @@ +package websocket + +import ( + "encoding/json" + "strings" + "testing" + + "github.com/breeze-rmm/agent/internal/wire" +) + +// oversizeResultBody builds a `result` body that marshals past the server's +// cap, shaped like the per-file arrays that actually cause this (a snapshot +// index, a software inventory, a filesystem walk). +func oversizeResultBody(entries int) map[string]any { + files := make([]map[string]any, 0, entries) + for i := 0; i < entries; i++ { + files = append(files, map[string]any{ + "sourcePath": `C:\Users\jdoe\AppData\Local\Cache\` + strings.Repeat("x", 64), + "backupPath": "snapshot-1/C_/Users/jdoe/AppData/Local/Cache/" + strings.Repeat("y", 64), + "checksum": strings.Repeat("a", 64), + "size": 4096 + i, + }) + } + return map[string]any{ + "id": "job-1", + "status": "completed", + "filesBackedUp": entries, + "snapshot": map[string]any{"id": "snapshot-1", "files": files}, + } +} + +// TestBoundResultFieldDropsOversizeBodyAndKeepsTerminalStatus is the direct +// #3001 regression: an oversize `result` must cost the BODY, never the terminal +// status. Before this the whole message was refused server-side and the job was +// reaped as stalled. +func TestBoundResultFieldDropsOversizeBodyAndKeepsTerminalStatus(t *testing.T) { + body := oversizeResultBody(4000) + encoded, err := json.Marshal(body) + if err != nil { + t.Fatalf("marshal fixture: %v", err) + } + if len(encoded) <= wire.MaxCommandResultBytes { + t.Fatalf("fixture is only %d bytes, not over the %d cap — the test would prove nothing", + len(encoded), wire.MaxCommandResultBytes) + } + + in := CommandResult{ + Type: "command_result", + CommandID: "cmd-3001", + Status: "completed", + ExitCode: 0, + Error: "3 files could not be read", + Result: body, + } + + out, changed := boundResultFieldForServer(in) + if !changed { + t.Fatal("an over-cap result body was left untouched; the server would reject the whole message") + } + if out.Status != "completed" || out.CommandID != "cmd-3001" || out.ExitCode != 0 { + t.Fatalf("terminal status was not preserved: %+v", out) + } + if out.Error != "3 files could not be read" { + t.Fatalf("error text was not preserved: %q", out.Error) + } + + marker, ok := out.Result.(map[string]any) + if !ok { + t.Fatalf("bounded result body is %T, want the marker map", out.Result) + } + if marker[resultOmittedMarker] != true { + t.Fatalf("marker key %q missing from %v", resultOmittedMarker, marker) + } + if marker["originalBytes"] != len(encoded) { + t.Fatalf("marker reports originalBytes=%v, want %d", marker["originalBytes"], len(encoded)) + } + if marker["limitBytes"] != wire.MaxCommandResultBytes { + t.Fatalf("marker reports limitBytes=%v, want %d", marker["limitBytes"], wire.MaxCommandResultBytes) + } + + // The point of the exercise: the bounded message is now deliverable. + reencoded, err := json.Marshal(out.Result) + if err != nil { + t.Fatalf("marshal bounded body: %v", err) + } + if len(reencoded) > wire.MaxCommandResultBytes { + t.Fatalf("bounded body is still %d bytes, over the %d cap", len(reencoded), wire.MaxCommandResultBytes) + } +} + +// TestBoundResultFieldLeavesInBudgetResultsAlone guards against the backstop +// becoming a silent data-loss path of its own: the overwhelming majority of +// command results are small and must reach the server byte-for-byte. +func TestBoundResultFieldLeavesInBudgetResultsAlone(t *testing.T) { + for _, tc := range []struct { + name string + body any + }{ + {"nil body", nil}, + {"small object", map[string]any{"filesBackedUp": 1200, "status": "completed"}}, + {"just under the cap", oversizeResultBody(1200)}, + } { + t.Run(tc.name, func(t *testing.T) { + if tc.body != nil { + encoded, err := json.Marshal(tc.body) + if err != nil { + t.Fatalf("marshal fixture: %v", err) + } + if len(encoded) > wire.MaxCommandResultBytes { + t.Fatalf("fixture is %d bytes, over the %d cap — it belongs in the oversize test", + len(encoded), wire.MaxCommandResultBytes) + } + } + in := CommandResult{CommandID: "c1", Status: "completed", Result: tc.body} + out, changed := boundResultFieldForServer(in) + if changed { + t.Fatal("an in-budget result body was rewritten; small results must pass through untouched") + } + if out.Result == nil && tc.body != nil { + t.Fatal("result body was dropped") + } + }) + } +} + +// TestBoundResultFieldHandlesUnmarshallableBody covers the branch where the +// body cannot be encoded at all. Marshalling it in SendResult would fail and +// the caller would return an error, losing the terminal status — the same +// outcome #3001 produced by a different route. +func TestBoundResultFieldHandlesUnmarshallableBody(t *testing.T) { + in := CommandResult{ + CommandID: "c1", + Status: "completed", + Result: make(chan int), // channels are not JSON-encodable + } + out, changed := boundResultFieldForServer(in) + if !changed { + t.Fatal("an unmarshallable body was left in place; the whole result would fail to encode") + } + if out.Status != "completed" { + t.Fatalf("terminal status lost: %+v", out) + } + if _, err := json.Marshal(out); err != nil { + t.Fatalf("bounded result still cannot be marshalled: %v", err) + } +} + +// TestSendResultBoundsOversizeBodyBeforeEnqueue proves the backstop is actually +// wired into the send path, not merely present. It asserts on the bytes that +// reach resultChan, which is what writePump puts on the wire. +func TestSendResultBoundsOversizeBodyBeforeEnqueue(t *testing.T) { + c := &Client{ + resultChan: make(chan outboundResult, 1), + done: make(chan struct{}), + } + + if err := c.SendResult(CommandResult{ + Type: "command_result", + CommandID: "cmd-3001", + Status: "completed", + Result: oversizeResultBody(4000), + }); err != nil { + t.Fatalf("SendResult: %v", err) + } + + select { + case queued := <-c.resultChan: + if len(queued.data) > wire.MaxCommandResultBytes { + t.Fatalf("enqueued frame is %d bytes; the server caps `result` at %d and would refuse it", + len(queued.data), wire.MaxCommandResultBytes) + } + var decoded struct { + CommandID string `json:"commandId"` + Status string `json:"status"` + Result map[string]any `json:"result"` + } + if err := json.Unmarshal(queued.data, &decoded); err != nil { + t.Fatalf("unmarshal enqueued frame: %v", err) + } + if decoded.Status != "completed" || decoded.CommandID != "cmd-3001" { + t.Fatalf("terminal status lost in the enqueued frame: %+v", decoded) + } + if decoded.Result[resultOmittedMarker] != true { + t.Fatalf("enqueued frame does not carry the omission marker: %v", decoded.Result) + } + default: + t.Fatal("SendResult enqueued nothing") + } +} diff --git a/agent/internal/wire/limits.go b/agent/internal/wire/limits.go new file mode 100644 index 000000000..ee3861fd7 --- /dev/null +++ b/agent/internal/wire/limits.go @@ -0,0 +1,59 @@ +// Package wire holds the size limits the SERVER enforces on agent→server +// messages. +// +// They live in their own leaf package, with no dependencies, because the two +// places that must respect them sit on opposite sides of the agent: the backup +// helper (cmd/breeze-backup) which BUILDS an oversize-capable result, and the +// websocket client (internal/websocket) which SENDS it. Importing +// internal/websocket from the helper just to read a number would drag gorilla +// and the whole connection machinery into a helper binary that never opens a +// socket. +// +// Everything here mirrors a server-side constant. A value that drifts from its +// mirror silently re-opens the class of bug described below, so each one is +// pinned by a test that asserts the literal. +package wire + +// MaxCommandResultBytes is the server's cap on the encoded `result` field of a +// command_result message. +// +// MIRRORS apps/api/src/routes/agents/schemas.ts `MAX_COMMAND_RESULT_BYTES`. +// Pinned by TestMaxCommandResultBytesMatchesServerSchema; the server side is +// pinned by schemas.commandResult.test.ts. Both assert the literal, so raising +// one alone reddens CI rather than quietly reintroducing #3001. +// +// WHY THIS IS THE LIMIT THAT MATTERS. It is the tightest bound anywhere on the +// result path, and by a wide margin: +// +// 1 MiB this — server-side Zod refine on `result` +// 16 MiB ipc.MaxMessageSize (helper→agent frame) +// 16 MiB websocket.maxMessageSize (agent's INBOUND read limit) +// 100 MiB the `ws` server's default maxPayload +// +// #3001: the backup helper's tiered degradation bounded against the 16 MiB IPC +// frame — the next hop, not the binding one — so it stayed inert while every +// backup over ~2,000 files was rejected by the server 16x below that budget. +// The rejection logged as a generic invalid-message server-side and as nothing +// at all agent-side, so a backup that had SUCCEEDED was reported to the user as +// stalled by the stale-backup reaper. Bound against the tightest limit in the +// whole chain, never merely the next one. +const MaxCommandResultBytes = 1024 * 1024 + +// CommandResultHeadroom is subtracted from MaxCommandResultBytes to get the +// budget an agent-side producer should actually target. +// +// It covers the gap between what Go measures and what the server measures. The +// server re-encodes with JSON.stringify AFTER JSON.parse, so the byte count it +// checks is not the byte count Go produced. Go's encoding/json is the more +// verbose of the two in every direction that matters — it HTML-escapes `<`, +// `>` and `&` to six-byte /& sequences and escapes U+2028 and +// U+2029, none of which JSON.stringify does — so Go's length is a conservative +// (>=) proxy for the server's on any object body. The headroom covers the +// residual: number re-formatting, and the case where a producer's JSON text is +// not an object at all and reaches the wire as a JSON *string*, where quoting +// and escaping make the server's measurement LARGER than the raw text. +const CommandResultHeadroom = 64 * 1024 + +// CommandResultBudget is the size an agent-side producer should keep its +// encoded result body under so the server accepts it. +const CommandResultBudget = MaxCommandResultBytes - CommandResultHeadroom diff --git a/agent/internal/wire/limits_test.go b/agent/internal/wire/limits_test.go new file mode 100644 index 000000000..960f90a5c --- /dev/null +++ b/agent/internal/wire/limits_test.go @@ -0,0 +1,75 @@ +package wire + +import ( + "os" + "path/filepath" + "regexp" + "strconv" + "strings" + "testing" +) + +// serverSchemaPath is the TypeScript file that declares the authoritative +// server-side cap. Relative to this package's directory. +const serverSchemaPath = "../../../apps/api/src/routes/agents/schemas.ts" + +// TestMaxCommandResultBytesMatchesServerSchema pins the Go mirror to the +// literal AND to the server's declaration. +// +// The literal assertion catches a local edit; parsing the TypeScript catches +// the dangerous direction — someone raising the server cap and never touching +// the agent, which leaves the agent bounding to a stale, tighter budget (merely +// wasteful) or, if lowered, to a stale looser one, which is exactly the #3001 +// failure: the agent believed it had 16 MiB of room while the server allowed +// 1 MiB, and every backup over ~2,000 files was silently rejected. +func TestMaxCommandResultBytesMatchesServerSchema(t *testing.T) { + if MaxCommandResultBytes != 1048576 { + t.Fatalf("MaxCommandResultBytes = %d, want 1048576; if the server cap really moved, update "+ + "apps/api/src/routes/agents/schemas.ts MAX_COMMAND_RESULT_BYTES in the SAME commit", + MaxCommandResultBytes) + } + + path := filepath.Clean(serverSchemaPath) + data, err := os.ReadFile(path) + if err != nil { + // A missing file is a failure, not a skip: this assertion is the only + // thing tying the two sides together, and a silently skipped + // cross-language pin is the same as no pin at all. + t.Fatalf("cannot read the server schema at %s to verify the mirrored cap: %v", path, err) + } + + re := regexp.MustCompile(`MAX_COMMAND_RESULT_BYTES\s*=\s*([0-9_]+)`) + m := re.FindSubmatch(data) + if m == nil { + t.Fatalf("no `MAX_COMMAND_RESULT_BYTES = ` declaration found in %s — if it was "+ + "renamed or moved, update serverSchemaPath and this pattern", path) + } + serverValue, err := strconv.Atoi(strings.ReplaceAll(string(m[1]), "_", "")) + if err != nil { + t.Fatalf("could not parse MAX_COMMAND_RESULT_BYTES value %q: %v", m[1], err) + } + if serverValue != MaxCommandResultBytes { + t.Fatalf("server MAX_COMMAND_RESULT_BYTES = %d but Go MaxCommandResultBytes = %d; the agent "+ + "bounds its payloads against the Go value, so a mismatch means results are being built "+ + "to a budget the server does not honour (issue #3001)", + serverValue, MaxCommandResultBytes) + } +} + +// TestCommandResultBudgetLeavesHeadroom guards the arithmetic rather than the +// numbers: the budget producers target must be strictly under the cap the +// server enforces, or the headroom that absorbs Go-vs-JS re-encoding +// differences is not actually there. +func TestCommandResultBudgetLeavesHeadroom(t *testing.T) { + if CommandResultBudget >= MaxCommandResultBytes { + t.Fatalf("CommandResultBudget (%d) must be strictly below MaxCommandResultBytes (%d)", + CommandResultBudget, MaxCommandResultBytes) + } + if CommandResultBudget <= 0 { + t.Fatalf("CommandResultBudget (%d) must be positive; headroom (%d) has swallowed the cap (%d)", + CommandResultBudget, CommandResultHeadroom, MaxCommandResultBytes) + } + if got := MaxCommandResultBytes - CommandResultBudget; got != CommandResultHeadroom { + t.Fatalf("budget/headroom/cap are inconsistent: cap-budget = %d, headroom = %d", got, CommandResultHeadroom) + } +} diff --git a/apps/api/src/routes/agentWs.ts b/apps/api/src/routes/agentWs.ts index 994842105..135397375 100644 --- a/apps/api/src/routes/agentWs.ts +++ b/apps/api/src/routes/agentWs.ts @@ -62,7 +62,11 @@ import { SW_INSTALL_COMMAND_ID_REGEX, } from '../services/softwareDeploymentResult'; import { PG_UUID_REGEX, UUID_REGEX } from '../utils/uuid'; -import { commandResultSchema as baseCommandResultSchema } from './agents/schemas'; +import { + commandResultSchema as baseCommandResultSchema, + commandResultResultByteLength, + MAX_COMMAND_RESULT_BYTES, +} from './agents/schemas'; import { commandResultHandlers, normalizeDiscoveryHosts } from '../services/commandResultHandlers'; /** Capabilities advertised to agents in the post-connect `connected` message. */ @@ -2285,11 +2289,45 @@ export function createAgentWsHandlers(agentId: string, preValidatedAgent: AgentD const parsed = agentMessageSchema.safeParse(message); if (!parsed.success) { - console.warn(`Invalid message from agent ${agentId}:`, parsed.error.issues); + // #3001: this branch used to be a bare console.warn carrying only the + // raw Zod issues. When it rejected a `command_result` — the terminal + // status of a job the server is actively waiting on — the operator + // saw no commandId, no size, no message type, and none of the + // downstream backup logs (those live in processCommandResult, past + // this return). The result read as having vanished in transit, and a + // succeeded backup was reaped as stalled 15 minutes later. + // + // A rejected command_result is a LOST TERMINAL STATUS, so it logs at + // error with everything needed to identify the job and the offending + // field; anything else stays a warning. + const rejectedType = typeof message?.type === 'string' ? message.type : 'unknown'; + const rejectedCommandId = typeof message?.commandId === 'string' + ? message.commandId + : undefined; + if (rejectedType === 'command_result') { + const resultBytes = commandResultResultByteLength(message?.result); + console.error( + `[AgentWs] REJECTED command_result from agent ${agentId} — the job will have no ` + + `terminal status and will be failed by a reaper. commandId=${rejectedCommandId ?? 'unknown'} ` + + `frameBytes=${data.length} resultBytes=${resultBytes ?? 'unencodable'} ` + + `resultLimitBytes=${MAX_COMMAND_RESULT_BYTES}:`, + parsed.error.issues + ); + } else { + console.warn( + `Invalid message from agent ${agentId} (type=${rejectedType}, frameBytes=${data.length}):`, + parsed.error.issues + ); + } ws.send(JSON.stringify({ type: 'error', code: 'INVALID_MESSAGE', message: 'Invalid message format', + // Echoed so the agent can attribute the rejection to the command it + // sent. Without it the agent sees an unattributable error frame and + // has nothing to log against the job. + messageType: rejectedType, + commandId: rejectedCommandId, details: parsed.error.issues })); return; diff --git a/apps/api/src/routes/agents/schemas.commandResult.test.ts b/apps/api/src/routes/agents/schemas.commandResult.test.ts new file mode 100644 index 000000000..ed5d80cfd --- /dev/null +++ b/apps/api/src/routes/agents/schemas.commandResult.test.ts @@ -0,0 +1,137 @@ +import { describe, it, expect } from 'vitest'; +import { readFileSync } from 'node:fs'; +import { resolve } from 'node:path'; +import { + commandResultSchema, + commandResultResultByteLength, + MAX_COMMAND_RESULT_BYTES, +} from './schemas'; + +// The Go mirror of the cap. Relative to this file's directory. +const GO_LIMITS_PATH = resolve(__dirname, '../../../../../agent/internal/wire/limits.go'); + +function validResult(overrides: Record = {}) { + return { + status: 'completed' as const, + exitCode: 0, + ...overrides, + }; +} + +/** + * Build a `result` value whose JSON.stringify output is exactly `bytes` long. + * A single ASCII string field lets the size be dialled precisely, which is what + * makes the boundary assertions meaningful rather than approximate. + */ +function resultOfExactBytes(bytes: number) { + // {"p":""} => 10 bytes of structure around the padding. + const structural = JSON.stringify({ p: '' }).length; + return { p: 'x'.repeat(bytes - structural) }; +} + +describe('command_result `result` size cap (#3001)', () => { + it('pins MAX_COMMAND_RESULT_BYTES to the value the agent mirrors', () => { + // The agent bounds its payloads against its own copy of this number. If + // this changes without agent/internal/wire/limits.go changing in the same + // commit, agents build results to a budget the server no longer honours — + // which is exactly how #3001 stayed invisible: the agent bounded against a + // 16 MiB IPC frame while the server enforced 1 MiB here. + expect(MAX_COMMAND_RESULT_BYTES).toBe(1_048_576); + }); + + it('matches the Go mirror in agent/internal/wire/limits.go', () => { + const source = readFileSync(GO_LIMITS_PATH, 'utf8'); + const declaration = source.match(/MaxCommandResultBytes\s*=\s*([0-9*\s]+)/)?.[1]; + if (declaration === undefined) { + throw new Error(`no MaxCommandResultBytes declaration found in ${GO_LIMITS_PATH}`); + } + + // The Go side writes it as an arithmetic literal (1024 * 1024). + const goValue = declaration + .split('*') + .map((part) => Number(part.trim())) + .reduce((a, b) => a * b, 1); + + expect(goValue).toBe(MAX_COMMAND_RESULT_BYTES); + }); + + it('accepts a result body exactly at the cap', () => { + const parsed = commandResultSchema.safeParse( + validResult({ result: resultOfExactBytes(MAX_COMMAND_RESULT_BYTES) }) + ); + expect(parsed.success).toBe(true); + }); + + it('rejects a result body one byte over the cap', () => { + const parsed = commandResultSchema.safeParse( + validResult({ result: resultOfExactBytes(MAX_COMMAND_RESULT_BYTES + 1) }) + ); + expect(parsed.success).toBe(false); + // The message must name the limit and the field, because the WS handler + // logs these issues verbatim and an operator reads them cold. + expect(JSON.stringify(parsed.success ? [] : parsed.error.issues)).toContain( + String(MAX_COMMAND_RESULT_BYTES) + ); + }); + + it('rejects a ~2 MB snapshot file index — the shape that reproduced #3001', () => { + // ~4,000 snapshot entries at ~522 B each is what a 4,000-file backup + // produced; the terminal result was refused here and the job was reaped as + // stalled. A 1,200-file run (~0.6 MB) passed, which is why the loss + // threshold sat between the two. + const files = Array.from({ length: 4000 }, (_, i) => ({ + sourcePath: `C:\\Users\\jdoe\\AppData\\Local\\Cache\\Cache_Data\\f_${i}${'x'.repeat(60)}`, + backupPath: `snapshot-1/C_/Users/jdoe/AppData/Local/Cache/Cache_Data/f_${i}${'y'.repeat(60)}`, + checksum: 'a'.repeat(64), + size: 4096 + i, + modTime: '2026-07-14T09:12:33Z', + })); + const body = { id: 'job-1', status: 'completed', snapshot: { id: 'snapshot-1', files } }; + + expect(commandResultResultByteLength(body)!).toBeGreaterThan(MAX_COMMAND_RESULT_BYTES); + expect(commandResultSchema.safeParse(validResult({ result: body })).success).toBe(false); + + // ...and the degraded form the agent now sends instead is accepted, with + // the terminal status intact. + const degraded = { ...body, snapshot: { id: 'snapshot-1', files: [] } }; + const parsed = commandResultSchema.safeParse(validResult({ result: degraded })); + expect(parsed.success).toBe(true); + expect(parsed.success && parsed.data.status).toBe('completed'); + }); + + it('allows an absent or null result body', () => { + expect(commandResultSchema.safeParse(validResult()).success).toBe(true); + expect(commandResultSchema.safeParse(validResult({ result: null })).success).toBe(true); + }); + + it('does not apply the result cap to stdout, which has its own 5 MB budget', () => { + // Guards against a fix that "unifies" the caps by tightening stdout: script + // output legitimately runs to megabytes and must not start being rejected. + const parsed = commandResultSchema.safeParse( + validResult({ stdout: 'x'.repeat(MAX_COMMAND_RESULT_BYTES + 1) }) + ); + expect(parsed.success).toBe(true); + }); +}); + +describe('commandResultResultByteLength', () => { + it('measures UTF-8 bytes, not UTF-16 code units', () => { + // The pre-#3097 WS copy measured with .length, accepting roughly 3x the + // intended budget for CJK-heavy output. Keep that fixed. + const body = { p: '日'.repeat(100) }; + expect(commandResultResultByteLength(body)).toBe(Buffer.byteLength(JSON.stringify(body), 'utf8')); + expect(commandResultResultByteLength(body)).toBeGreaterThan(JSON.stringify(body).length); + }); + + it('returns 0 for absent bodies', () => { + expect(commandResultResultByteLength(undefined)).toBe(0); + expect(commandResultResultByteLength(null)).toBe(0); + }); + + it('returns null for a body that cannot be serialised, and the schema rejects it', () => { + const cyclic: Record = {}; + cyclic.self = cyclic; + expect(commandResultResultByteLength(cyclic)).toBeNull(); + expect(commandResultSchema.safeParse(validResult({ result: cyclic })).success).toBe(false); + }); +}); diff --git a/apps/api/src/routes/agents/schemas.ts b/apps/api/src/routes/agents/schemas.ts index 9c72299cd..4eec158f6 100644 --- a/apps/api/src/routes/agents/schemas.ts +++ b/apps/api/src/routes/agents/schemas.ts @@ -278,6 +278,53 @@ export const processSampleSchema = z.object({ // Commands // ============================================ +/** + * MAX_COMMAND_RESULT_BYTES bounds the `result` field of an agent command + * result. It is the TIGHTEST limit anywhere on the agent→server result path — + * tighter than the 16 MiB agent IPC frame, the agent's 16 MiB WS read limit and + * the `ws` server's 100 MiB default `maxPayload` — so it, not any of those, is + * the limit an agent has to bound its payload against. + * + * That is not obvious from the agent side, and #3001 is what it costs when it + * is missed: the backup helper emitted a BackupJob body carrying one + * `snapshot.files` entry per backed-up file (~522 B each), the agent put that + * body in `result`, and every backup over ~2,000 files was rejected here. The + * agent's own tiered degradation was bounding against the 16 MiB IPC frame — + * 16x too loose — so it never fired, nothing logged on either side, and the job + * sat `running` until the stale-backup reaper falsely failed a backup that had + * in fact succeeded. + * + * MIRRORED IN GO as `wire.MaxCommandResultBytes` + * (agent/internal/wire/limits.go). The two are pinned equal by + * `schemas.commandResult.test.ts` here and `TestMaxCommandResultBytesMatchesServerSchema` + * there — both assert the literal 1048576, so raising one alone reddens CI on + * both sides rather than silently re-opening #3001. + * + * Deliberately NOT raised to match the 5 MB `stdout`/`stderr` caps: a larger cap + * only moves the cliff (a 100k-file snapshot index is ~52 MB and fits no sane + * cap), and the agent-side bound is the actual fix. Changing this value is a + * one-line, reversible decision — but it must be changed on BOTH sides. + */ +export const MAX_COMMAND_RESULT_BYTES = 1_048_576; + +/** + * commandResultResultByteLength returns the encoded size the `result` field + * will occupy, or null when the value cannot be serialised at all (a cycle, or + * a throwing toJSON). Exported so the WS layer can report the measured size in + * its rejection log instead of leaving an operator to guess why a result was + * refused. + */ +export function commandResultResultByteLength(val: unknown): number | null { + if (val === undefined || val === null) return 0; + try { + const encoded = JSON.stringify(val); + if (encoded === undefined) return 0; + return Buffer.byteLength(encoded, 'utf8'); + } catch { + return null; + } +} + export const commandResultSchema = z.object({ status: z.enum(['completed', 'failed', 'timeout']), exitCode: z.number().int().optional(), @@ -291,10 +338,14 @@ export const commandResultSchema = z.object({ error: z.string().max(10_000).optional(), result: z.any().optional().refine( (val) => { - if (val === undefined || val === null) return true; - try { return Buffer.byteLength(JSON.stringify(val), 'utf8') <= 1_048_576; } catch { return false; } + const size = commandResultResultByteLength(val); + return size !== null && size <= MAX_COMMAND_RESULT_BYTES; }, - { message: 'Command result payload exceeds 1 MB limit' } + { + message: + `Command result payload exceeds the ${MAX_COMMAND_RESULT_BYTES}-byte \`result\` limit ` + + '(the agent must degrade the payload before sending — see #3001)' + } ) }); From 88a32ea495822860b84c6d58c6d294f85f11e4de Mon Sep 17 00:00:00 2001 From: Todd Hebebrand Date: Sat, 8 Aug 2026 00:22:47 -0600 Subject: [PATCH 2/3] fix(agent/ws): bound the generic result backstop against the budget, not the bare cap MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review finding on #3267. boundResultFieldForServer and the SendResult short-circuit compared against wire.MaxCommandResultBytes, so the one guard that protects every NON-backup command type ran with no margin for the server's JSON.stringify re-measurement. A body landing in the 64 KiB band below the cap could pass here and still be refused on arrival — the exact silent loss this PR closes, for the commands with no producer-side bounding. Both comparisons now use wire.CommandResultBudget; the bare cap is kept only for reporting the server's contract in logs and the omission marker. TestBoundResultFieldUsesTheBudgetNotTheBareCap pins the choice. Co-Authored-By: Claude Opus 4.8 (1M context) --- agent/internal/websocket/client.go | 19 ++++-- .../websocket/result_server_cap_test.go | 60 +++++++++++++++---- agent/internal/wire/limits.go | 39 +++++++----- 3 files changed, 87 insertions(+), 31 deletions(-) diff --git a/agent/internal/websocket/client.go b/agent/internal/websocket/client.go index bc4ee17ae..c3f6d07cd 100644 --- a/agent/internal/websocket/client.go +++ b/agent/internal/websocket/client.go @@ -750,10 +750,11 @@ func (c *Client) SendResult(result CommandResult) error { if err != nil { return fmt.Errorf("failed to marshal result: %w", err) } - if len(data) > wire.MaxCommandResultBytes { + if len(data) > wire.CommandResultBudget { // Only pay for the precise per-field measurement once the whole frame - // is over the cap — the `result` field is a subset of it, so a frame - // under the cap cannot contain an oversize field. + // is over the budget — the `result` field's encoded bytes are a subset + // of the frame's, so a frame under the budget cannot contain a field + // over it. if bounded, ok := boundResultFieldForServer(result); ok { result = bounded data, err = json.Marshal(result) @@ -816,14 +817,22 @@ func boundResultFieldForServer(result CommandResult) (CommandResult, bool) { } return result, true } - if len(encoded) <= wire.MaxCommandResultBytes { + // Measured against the BUDGET, not the bare cap. The server does not check + // these bytes: it re-encodes with JSON.stringify after JSON.parse and checks + // the result of that, so Go's count is a close proxy but not the same + // number. Spending the headroom here is nearly free — it drops a body only + // in the narrow band just under the cap, and the terminal status still + // lands — whereas being wrong by one byte costs the entire message. For + // every command type other than backup this is the only guard there is. + if len(encoded) <= wire.CommandResultBudget { return result, false } - log.Error("command result body exceeds the server's result cap and was dropped to preserve the terminal status", + log.Error("command result body exceeds the server's result budget and was dropped to preserve the terminal status", "commandId", result.CommandID, "status", result.Status, "resultBytes", len(encoded), + "budgetBytes", wire.CommandResultBudget, "limitBytes", wire.MaxCommandResultBytes, ) result.Result = map[string]any{ diff --git a/agent/internal/websocket/result_server_cap_test.go b/agent/internal/websocket/result_server_cap_test.go index 5c3758c03..5418866e8 100644 --- a/agent/internal/websocket/result_server_cap_test.go +++ b/agent/internal/websocket/result_server_cap_test.go @@ -39,9 +39,9 @@ func TestBoundResultFieldDropsOversizeBodyAndKeepsTerminalStatus(t *testing.T) { if err != nil { t.Fatalf("marshal fixture: %v", err) } - if len(encoded) <= wire.MaxCommandResultBytes { - t.Fatalf("fixture is only %d bytes, not over the %d cap — the test would prove nothing", - len(encoded), wire.MaxCommandResultBytes) + if len(encoded) <= wire.CommandResultBudget { + t.Fatalf("fixture is only %d bytes, not over the %d budget — the test would prove nothing", + len(encoded), wire.CommandResultBudget) } in := CommandResult{ @@ -83,8 +83,8 @@ func TestBoundResultFieldDropsOversizeBodyAndKeepsTerminalStatus(t *testing.T) { if err != nil { t.Fatalf("marshal bounded body: %v", err) } - if len(reencoded) > wire.MaxCommandResultBytes { - t.Fatalf("bounded body is still %d bytes, over the %d cap", len(reencoded), wire.MaxCommandResultBytes) + if len(reencoded) > wire.CommandResultBudget { + t.Fatalf("bounded body is still %d bytes, over the %d budget", len(reencoded), wire.CommandResultBudget) } } @@ -98,7 +98,7 @@ func TestBoundResultFieldLeavesInBudgetResultsAlone(t *testing.T) { }{ {"nil body", nil}, {"small object", map[string]any{"filesBackedUp": 1200, "status": "completed"}}, - {"just under the cap", oversizeResultBody(1200)}, + {"just under the budget", oversizeResultBody(1200)}, } { t.Run(tc.name, func(t *testing.T) { if tc.body != nil { @@ -106,9 +106,9 @@ func TestBoundResultFieldLeavesInBudgetResultsAlone(t *testing.T) { if err != nil { t.Fatalf("marshal fixture: %v", err) } - if len(encoded) > wire.MaxCommandResultBytes { - t.Fatalf("fixture is %d bytes, over the %d cap — it belongs in the oversize test", - len(encoded), wire.MaxCommandResultBytes) + if len(encoded) > wire.CommandResultBudget { + t.Fatalf("fixture is %d bytes, over the %d budget — it belongs in the oversize test", + len(encoded), wire.CommandResultBudget) } } in := CommandResult{CommandID: "c1", Status: "completed", Result: tc.body} @@ -123,6 +123,42 @@ func TestBoundResultFieldLeavesInBudgetResultsAlone(t *testing.T) { } } +// TestBoundResultFieldUsesTheBudgetNotTheBareCap pins the margin. +// +// A body sitting in the band between the budget and the cap is the case where +// Go's byte count and the server's JSON.stringify re-measurement can disagree. +// Comparing against wire.MaxCommandResultBytes here would let such a body +// through on a coin-flip, which for every non-backup command type — the ones +// with no producer-side bounding at all — means the whole message is refused +// and the terminal status is lost. That is #3001 exactly. +func TestBoundResultFieldUsesTheBudgetNotTheBareCap(t *testing.T) { + // {"p":""} — 10 bytes of structure around the padding, sized to + // land midway between the budget and the cap. + target := wire.CommandResultBudget + wire.CommandResultHeadroom/2 + body := map[string]any{"p": strings.Repeat("x", target-10)} + + encoded, err := json.Marshal(body) + if err != nil { + t.Fatalf("marshal fixture: %v", err) + } + if len(encoded) <= wire.CommandResultBudget || len(encoded) > wire.MaxCommandResultBytes { + t.Fatalf("fixture is %d bytes; it must land strictly between the budget (%d) and the cap (%d)", + len(encoded), wire.CommandResultBudget, wire.MaxCommandResultBytes) + } + + out, changed := boundResultFieldForServer(CommandResult{ + CommandID: "c1", Status: "completed", Result: body, + }) + if !changed { + t.Fatalf("a %d-byte body was left in place; the backstop is comparing against the bare cap (%d) "+ + "instead of the budget (%d), and has no margin for the server's re-encoding", + len(encoded), wire.MaxCommandResultBytes, wire.CommandResultBudget) + } + if out.Status != "completed" { + t.Fatalf("terminal status lost: %+v", out) + } +} + // TestBoundResultFieldHandlesUnmarshallableBody covers the branch where the // body cannot be encoded at all. Marshalling it in SendResult would fail and // the caller would return an error, losing the terminal status — the same @@ -165,9 +201,9 @@ func TestSendResultBoundsOversizeBodyBeforeEnqueue(t *testing.T) { select { case queued := <-c.resultChan: - if len(queued.data) > wire.MaxCommandResultBytes { - t.Fatalf("enqueued frame is %d bytes; the server caps `result` at %d and would refuse it", - len(queued.data), wire.MaxCommandResultBytes) + if len(queued.data) > wire.CommandResultBudget { + t.Fatalf("enqueued frame is %d bytes, over the %d budget; the server caps `result` at %d", + len(queued.data), wire.CommandResultBudget, wire.MaxCommandResultBytes) } var decoded struct { CommandID string `json:"commandId"` diff --git a/agent/internal/wire/limits.go b/agent/internal/wire/limits.go index ee3861fd7..52393bf71 100644 --- a/agent/internal/wire/limits.go +++ b/agent/internal/wire/limits.go @@ -40,20 +40,31 @@ package wire const MaxCommandResultBytes = 1024 * 1024 // CommandResultHeadroom is subtracted from MaxCommandResultBytes to get the -// budget an agent-side producer should actually target. -// -// It covers the gap between what Go measures and what the server measures. The -// server re-encodes with JSON.stringify AFTER JSON.parse, so the byte count it -// checks is not the byte count Go produced. Go's encoding/json is the more -// verbose of the two in every direction that matters — it HTML-escapes `<`, -// `>` and `&` to six-byte /& sequences and escapes U+2028 and -// U+2029, none of which JSON.stringify does — so Go's length is a conservative -// (>=) proxy for the server's on any object body. The headroom covers the -// residual: number re-formatting, and the case where a producer's JSON text is -// not an object at all and reaches the wire as a JSON *string*, where quoting -// and escaping make the server's measurement LARGER than the raw text. +// budget agent-side code should actually target. +// +// It exists because NOBODY on this side measures the same bytes the server +// does. The server re-encodes with JSON.stringify AFTER JSON.parse and checks +// the length of that, so every agent-side measurement is a proxy: +// +// - The backup helper measures the RAW JSON TEXT it produced (len(Stdout)), +// before that text is even parsed. When the text is not an object it +// reaches the wire as a JSON *string*, and the quoting and escaping make +// the server's measurement LARGER than the text — the one direction where +// the proxy under-reports. +// - The websocket client measures Go's encoding of the value. Go's +// encoding/json is the more verbose of the two in every direction that +// matters — it HTML-escapes `<`, `>` and `&` to six-byte / +// & sequences and escapes U+2028/U+2029, none of which JSON.stringify +// does — so this proxy is conservative except for number re-formatting. +// +// Rather than reason about which proxy is safe where, every producer targets +// the budget. Overshooting the margin costs a degraded body in a narrow band +// below the cap; undershooting it costs the whole message, which is the failure +// this package exists to prevent. const CommandResultHeadroom = 64 * 1024 -// CommandResultBudget is the size an agent-side producer should keep its -// encoded result body under so the server accepts it. +// CommandResultBudget is the size agent-side code should keep its encoded +// result body under so the server accepts it. Prefer this over +// MaxCommandResultBytes at every comparison site; the bare cap is for reporting +// the server's contract in logs and markers, not for deciding whether to send. const CommandResultBudget = MaxCommandResultBytes - CommandResultHeadroom From 16c9132b546612c9cbc4a51654e219eeb4011212 Mon Sep 17 00:00:00 2001 From: Todd Hebebrand Date: Sat, 8 Aug 2026 10:02:46 -0600 Subject: [PATCH 3/3] fix(backup): name the real limit in operator-facing warnings; repair the unencodable-body path MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review round on #3267. 1. The five PERSISTED warnings still hardcoded the IPC limit — the same bug this PR fixes in the log line, one layer further out. These strings land in result.warning -> backup_jobs.errorLog -> the UI, so after the binding limit moved to the 1 MiB server cap the headline 4,000-file repro would have told a customer that a 2 MB result exceeded a 16 MiB limit; oversizeFailureResult printed the actual size next to the limit it supposedly exceeded, making the contradiction visible in one line. The limit is now a value (deliveryLimit) threaded into every tier, so the warning, the failure text and the log line cannot tell different stories. It carries the enforced BUDGET separately from the enforcing party's CAP: degradation trips at 983,040 while the server allows 1,048,576, so reporting the cap as "exceeded" described a 1,000,000-byte payload as overflowing a limit it was under. Warnings name the budget; the log carries both. Five existing tests asserted the literal "IPC limit" as a proxy for "the tier explained itself", which had welded the suite to one specific limit being the trigger forever. They now assert the shared limitExceededPhrase. 2. The unencodable-body recovery in the WS backstop was dead code. SendResult marshalled the whole result first and returned on error, so boundResultFieldForServer's marshal-error branch was unreachable from the wired path and its unit test asserted a repair that never ran. A NaN in any nested map still lost the terminal status. SendResult now substitutes the marker and retries, and still errors when there is no body to drop. 3. commandResultResultByteLength re-serialised a just-parsed object with no size guard, on the one branch guaranteed to see unbounded input (no maxPayload is set; pre-fix agents send 64 MB results). Above 8 MB the frame size is reported instead — this repo has event-loop-stall form (#3236). 4. frameBytes used String.length (UTF-16 code units) — the exact confusion this PR's own test pins — and the echoed messageType/commandId came off an unvalidated message unbounded. Now Buffer.byteLength and clamped. Tests: the rejection frame is pinned from BOTH sides (Vitest asserts the shape the Go parser reads; a Go test asserts the field names against agentWs.ts and that the readPump error branch still precedes the id-less skip, the ordering the fix depends on). Plus IPC-frame attribution, budget-vs-cap reporting, the persisted-warning wording, and the wired NaN path. Both cross-language pin regexes are anchored on the declaration keyword so a doc comment cannot retarget them onto prose. Co-Authored-By: Claude Opus 4.8 (1M context) --- agent/cmd/breeze-backup/result_bounds.go | 163 ++++++++++----- .../result_bounds_server_cap_test.go | 187 ++++++++++++++++-- agent/cmd/breeze-backup/result_bounds_test.go | 4 +- .../result_warning_preservation_test.go | 6 +- agent/internal/websocket/client.go | 29 ++- .../websocket/result_server_cap_test.go | 65 ++++++ .../websocket/server_error_frame_test.go | 120 +++++++++++ agent/internal/wire/limits_test.go | 8 +- .../src/routes/agentWs.rejectionFrame.test.ts | 142 +++++++++++++ apps/api/src/routes/agentWs.ts | 163 +++++++++++---- .../agents/schemas.commandResult.test.ts | 7 +- 11 files changed, 774 insertions(+), 120 deletions(-) create mode 100644 agent/internal/websocket/server_error_frame_test.go create mode 100644 apps/api/src/routes/agentWs.rejectionFrame.test.ts diff --git a/agent/cmd/breeze-backup/result_bounds.go b/agent/cmd/breeze-backup/result_bounds.go index b6f9c765b..350c42e28 100644 --- a/agent/cmd/breeze-backup/result_bounds.go +++ b/agent/cmd/breeze-backup/result_bounds.go @@ -174,7 +174,7 @@ const bulkFieldThreshold = 4 * 1024 // was already within budget. It is non-empty exactly when the caller should log // loudly — a degraded result is a real (if survivable) loss of detail. func fitBackupResultForDelivery(result backupipc.BackupCommandResult) (backupipc.BackupCommandResult, string) { - fitted, notes, _, _ := fitBackupResult(result) + fitted, notes, _ := fitBackupResult(result) return fitted, notes } @@ -189,14 +189,16 @@ func fitBackupResultForDelivery(result backupipc.BackupCommandResult) (backupipc // the problem. The trigger is therefore tracked as the tiers run rather than // assumed at the end. func fitBackupResult(result backupipc.BackupCommandResult) ( - fitted backupipc.BackupCommandResult, notesText, limitName string, limitBytes int, + fitted backupipc.BackupCommandResult, notesText string, limit deliveryLimit, ) { var notes []string // Captured BEFORE any tier runs: once the tiers have shrunk the payload the // binding limit is no longer observable, and reporting a limit derived from - // the already-degraded result is how the wrong one gets named. - limitName, limitBytes = exceededLimit(result) + // the already-degraded result is how the wrong one gets named. Every tier + // below is handed this value so the warning it persists names the same + // limit the log line does. + limit = exceededLimit(result) // Tier 1 — always applied, and cheap (no JSON parse). Bounding the failure // detail BEFORE marshalling is the primary fix; the tiers below are the @@ -209,7 +211,7 @@ func fitBackupResult(result backupipc.BackupCommandResult) ( notes = append(notes, fmt.Sprintf("stderr truncated (%d bytes dropped)", dropped)) } if fits(result) { - return finishBounding(result, notes, limitName, limitBytes) + return finishBounding(result, notes, limit) } // Only reached when the result is actually oversize, so the ordinary path @@ -217,7 +219,7 @@ func fitBackupResult(result backupipc.BackupCommandResult) ( obj, isObject := decodeStdoutObject(result.Stdout) if !isObject { notes = append(notes, "result body could not be summarised and was replaced with an oversize failure") - return finishBounding(oversizeFailureResult(result), notes, limitName, limitBytes) + return finishBounding(oversizeFailureResult(result, limit), notes, limit) } // Tier 2 — cap the warning text, empty the per-file snapshot index, and @@ -233,33 +235,33 @@ func fitBackupResult(result backupipc.BackupCommandResult) ( if dropped := boundObjectWarning(obj); dropped > 0 { notes = append(notes, fmt.Sprintf("warning truncated (%d bytes dropped)", dropped)) } - if entries, ok := emptySnapshotFiles(obj); ok { + if entries, ok := emptySnapshotFiles(obj, limit); ok { notes = append(notes, fmt.Sprintf("snapshot file index dropped (%d entries)", entries)) } - for _, key := range dropBulkFields(obj) { + for _, key := range dropBulkFields(obj, limit) { notes = append(notes, fmt.Sprintf("%s dropped (bulk field)", key)) } result.Stdout = encodeStdoutObject(obj, result.Stdout) if fits(result) { - return finishBounding(result, notes, limitName, limitBytes) + return finishBounding(result, notes, limit) } // Tier 3 — scalar fields plus the snapshot identity only. - if reduced, ok := reduceToScalars(result.Stdout); ok { + if reduced, ok := reduceToScalars(result.Stdout, limit); ok { result.Stdout = reduced notes = append(notes, "result reduced to summary scalars only") if fits(result) { - return finishBounding(result, notes, limitName, limitBytes) + return finishBounding(result, notes, limit) } } // Tier 4 — last resort. Every field is bounded by construction, so this // always fits: a terminal status must land even when nothing else can. notes = append(notes, "result replaced with a minimal terminal status") - result.Stdout = lastResortStdout(result.Stdout) + result.Stdout = lastResortStdout(result.Stdout, limit) result.Stderr, _ = truncateText(result.Stderr, maxLastResortFieldBytes) result.CommandID, _ = truncateText(result.CommandID, maxLastResortFieldBytes) - return finishBounding(result, notes, limitName, limitBytes) + return finishBounding(result, notes, limit) } // finishBounding assembles fitBackupResult's return, attributing the @@ -270,13 +272,13 @@ func fitBackupResult(result backupipc.BackupCommandResult) ( // were always going to fit the wire comfortably. Those are the runs the old log // line mislabelled as frame overflows. func finishBounding( - result backupipc.BackupCommandResult, notes []string, limitName string, limitBytes int, -) (backupipc.BackupCommandResult, string, string, int) { + result backupipc.BackupCommandResult, notes []string, limit deliveryLimit, +) (backupipc.BackupCommandResult, string, deliveryLimit) { notesText := joinNotes(notes) - if notesText != "" && limitName == "" { - limitName, limitBytes = limitResultText, maxResultTextBytes + if notesText != "" && !limit.fired() { + limit = limitResultText } - return result, notesText, limitName, limitBytes + return result, notesText, limit } // sendBackupResult sends a terminal backup_result envelope, bounding the @@ -288,7 +290,7 @@ func finishBounding( // log-shipping threshold and therefore on the server. func sendBackupResult(conn *ipc.Conn, envelopeID string, result backupipc.BackupCommandResult) error { log := logging.L("backup") - fitted, degraded, limitName, limitBytes := fitBackupResult(result) + fitted, degraded, limit := fitBackupResult(result) if degraded != "" { // Sizes are reported from the raw fields rather than a full marshal of // the original: re-marshalling tens of megabytes just to populate a log @@ -299,8 +301,13 @@ func sendBackupResult(conn *ipc.Conn, envelopeID string, result backupipc.Backup log.Error("backup result payload was degraded to fit a delivery limit", "commandId", result.CommandID, "degraded", degraded, - "limitName", limitName, - "limitBytes", limitBytes, + "limitName", limit.name, + // Both numbers: budgetBytes is the threshold actually crossed, + // limitBytes is what the enforcing party allows. Reporting only the + // latter would describe a 1,000,000-byte payload as having exceeded + // the 1,048,576-byte server cap, which is false. + "budgetBytes", limit.budget, + "limitBytes", limit.cap, "originalStdoutBytes", len(result.Stdout), "originalStderrBytes", len(result.Stderr), "sentBytes", marshalledSize(fitted), @@ -323,8 +330,7 @@ func sendBackupResult(conn *ipc.Conn, envelopeID string, result backupipc.Backup // fits reports whether result is within every delivery limit on the path to // the server. func fits(result backupipc.BackupCommandResult) bool { - name, _ := exceededLimit(result) - return name == "" + return !exceededLimit(result).fired() } // exceededLimit names the first delivery limit result's payload breaches, and @@ -339,14 +345,14 @@ func fits(result backupipc.BackupCommandResult) bool { // The raw text fields are measured before marshalling as a cheap lower bound — // JSON string encoding never shrinks its input — so an oversize result is // rejected without marshalling tens of megabytes on the endpoint. -func exceededLimit(result backupipc.BackupCommandResult) (string, int) { +func exceededLimit(result backupipc.BackupCommandResult) deliveryLimit { // Tightest first. Stdout alone, because Stdout is what becomes the server's // `result` field; see serverResultBudget. if len(result.Stdout) > serverResultBudget { - return limitServerResult, wire.MaxCommandResultBytes + return limitServerResult } if len(result.Stdout)+len(result.Stderr)+len(result.CommandID) > resultPayloadBudget { - return limitIPCFrame, ipc.MaxMessageSize + return limitIPCFrame } size := marshalledSize(result) // A result that cannot be marshalled at all is not "fitting" — treating @@ -355,17 +361,76 @@ func exceededLimit(result backupipc.BackupCommandResult) (string, int) { // (BackupCommandResult is only strings/bool/int64, so this is unreachable // today; it is guarded so it stays unreachable if the type grows a field.) if size < 0 || size > resultPayloadBudget { - return limitIPCFrame, ipc.MaxMessageSize + return limitIPCFrame } - return "", 0 + return deliveryLimit{} } -// Limit names used in the degradation log line. They are the identifiers an -// operator can grep for in this repo, not prose. -const ( - limitServerResult = "server command_result `result` cap (wire.MaxCommandResultBytes)" - limitIPCFrame = "agent IPC frame (ipc.MaxMessageSize)" - limitResultText = "free-text field cap (maxResultTextBytes)" +// deliveryLimit identifies the limit that forced a degradation, in the three +// forms this file needs: a grep-able identifier for structured logs, a short +// phrase for the operator-facing warning persisted to backup_jobs.errorLog, and +// the two byte counts. +// +// BUDGET AND CAP ARE TRACKED SEPARATELY because they are not the same number, +// and conflating them produces the very class of false statement this file's +// #3001 fix set out to kill. The server's cap is 1,048,576 but degradation +// trips at serverResultBudget (983,040), so reporting the cap as the thing that +// was "exceeded" tells an operator a 1,000,000-byte payload overflowed a limit +// it was comfortably under. The warning text names the budget — the threshold +// actually crossed — and the structured log carries both. +type deliveryLimit struct { + name string // grep-able identifier, for structured logs + label string // short operator-facing phrase, for persisted warnings + budget int // the threshold this code enforces; what was actually exceeded + cap int // the limit the enforcing party imposes; == budget when no headroom is held +} + +// fired reports whether a limit was breached at all. +func (l deliveryLimit) fired() bool { return l.name != "" } + +// limitExceededPhrase is the lead-in every tier's operator-facing note shares. +// +// It exists so a test can assert "this tier explained why it degraded" WITHOUT +// pinning which limit fired. Several tests previously asserted the literal +// "IPC limit" as that proxy, which quietly welded the suite to one specific +// limit being the trigger forever — so when the binding limit moved to the +// server cap, the tests kept passing on wording that had become false and would +// have failed on wording that had become true. Assert the shared phrase, or the +// specific limit via deliveryLimit.describe(); never a bare limit name. +const limitExceededPhrase = "exceeded the " + +// describe renders the operator-facing form: the threshold that was crossed and +// what imposes it. This is what reaches the customer in the job's error log, so +// it names a real, honest number — never a limit the payload was under. +func (l deliveryLimit) describe() string { + return fmt.Sprintf("%d byte %s", l.budget, l.label) +} + +// The limits, in the order exceededLimit checks them. +// +// These deliberately are NOT consts: each carries the budget it is enforced at, +// which is derived arithmetic, and the operator-facing label is part of the +// value rather than duplicated at five call sites — the duplication being how +// the five warning strings all came to hardcode the IPC limit. +var ( + limitServerResult = deliveryLimit{ + name: "server command_result `result` cap (wire.MaxCommandResultBytes)", + label: "server result budget", + budget: serverResultBudget, + cap: wire.MaxCommandResultBytes, + } + limitIPCFrame = deliveryLimit{ + name: "agent IPC frame (ipc.MaxMessageSize)", + label: "agent IPC budget", + budget: resultPayloadBudget, + cap: ipc.MaxMessageSize, + } + limitResultText = deliveryLimit{ + name: "free-text field cap (maxResultTextBytes)", + label: "free-text field cap", + budget: maxResultTextBytes, + cap: maxResultTextBytes, + } ) // marshalledSize returns the marshalled byte length of result, or -1 when it @@ -447,7 +512,7 @@ func boundObjectWarning(obj map[string]json.RawMessage) int { // the drop in the result's `warning`, so it is visible server-side (the warning // is persisted to the job's errorLog) rather than only in an endpoint log line. // Reports the number of entries dropped and whether anything was dropped. -func emptySnapshotFiles(obj map[string]json.RawMessage) (int, bool) { +func emptySnapshotFiles(obj map[string]json.RawMessage, limit deliveryLimit) (int, bool) { rawSnap, present := obj["snapshot"] if !present { return 0, false @@ -471,8 +536,8 @@ func emptySnapshotFiles(obj map[string]json.RawMessage) (int, bool) { } obj["snapshot"] = rebuilt obj["warning"] = mustRawString(appendResultWarning(obj["warning"], fmt.Sprintf( - "snapshot file index omitted (%d entries): the result exceeded the %d byte agent IPC limit, so per-file restore browsing is unavailable for this snapshot", - len(files), ipc.MaxMessageSize))) + "snapshot file index omitted (%d entries): the result exceeded the %s, so per-file restore browsing is unavailable for this snapshot", + len(files), limit.describe()))) return len(files), true } @@ -482,7 +547,7 @@ func emptySnapshotFiles(obj map[string]json.RawMessage) (int, bool) { // Snapshot.Files is. `snapshot` is exempt: emptySnapshotFiles already handled // its bulk and the rest of it is the identity the server needs. Returns the // keys dropped, sorted for a deterministic note, and records them in `warning`. -func dropBulkFields(obj map[string]json.RawMessage) []string { +func dropBulkFields(obj map[string]json.RawMessage, limit deliveryLimit) []string { var dropped []string for key, raw := range obj { if key == "snapshot" || len(raw) <= bulkFieldThreshold { @@ -499,8 +564,8 @@ func dropBulkFields(obj map[string]json.RawMessage) []string { } sort.Strings(dropped) obj["warning"] = mustRawString(appendResultWarning(obj["warning"], fmt.Sprintf( - "detail field(s) omitted (%s): the result exceeded the %d byte agent IPC limit", - strings.Join(dropped, ", "), ipc.MaxMessageSize))) + "detail field(s) omitted (%s): the result exceeded the %s", + strings.Join(dropped, ", "), limit.describe()))) return dropped } @@ -526,7 +591,7 @@ func isJSONContainer(raw json.RawMessage) bool { // what stops this tier from silently zeroing a counter the server reads // (filesRestored, filesFailed, errorCount, …) on a command shape it never // anticipated. -func reduceToScalars(stdout string) (string, bool) { +func reduceToScalars(stdout string, limit deliveryLimit) (string, bool) { obj, ok := decodeStdoutObject(stdout) if !ok { return stdout, false @@ -576,8 +641,8 @@ func reduceToScalars(stdout string) (string, bool) { if len(containers) > 0 { sort.Strings(containers) kept["warning"] = mustRawString(appendResultWarning(kept["warning"], fmt.Sprintf( - "detail field(s) omitted (%s): the result exceeded the %d byte agent IPC limit", - strings.Join(containers, ", "), ipc.MaxMessageSize))) + "detail field(s) omitted (%s): the result exceeded the %s", + strings.Join(containers, ", "), limit.describe()))) } return encodeStdoutObject(kept, stdout), true } @@ -587,15 +652,15 @@ func reduceToScalars(stdout string) (string, bool) { // failure. A terminal status still lands — the point of #3001 — but an empty // body is never handed back under Success: true, where the server would read it // as a complete, empty answer. -func oversizeFailureResult(result backupipc.BackupCommandResult) backupipc.BackupCommandResult { +func oversizeFailureResult(result backupipc.BackupCommandResult, limit deliveryLimit) backupipc.BackupCommandResult { // Size is captured before the fields are cleared — reporting it after would // print the size of the replacement, not of what was dropped. original := len(result.Stdout) + len(result.Stderr) result.Stdout = "" result.Success = false result.Stderr = fmt.Sprintf( - "backup helper result exceeded the %d byte agent IPC limit (%d bytes) and could not be summarised; the command may have succeeded but its output could not be delivered", - ipc.MaxMessageSize, original) + "backup helper result exceeded the %s (%d bytes) and could not be summarised; the command may have succeeded but its output could not be delivered", + limit.describe(), original) result.CommandID, _ = truncateText(result.CommandID, maxLastResortFieldBytes) return result } @@ -612,7 +677,7 @@ func oversizeFailureResult(result backupipc.BackupCommandResult) backupipc.Backu // clean-looking one on exactly the runs most likely to be degraded, since a // large run is what reaches this tier at all. The tier may truncate the warning // and must append its own note to it; it must never substitute for it. -func lastResortStdout(stdout string) string { +func lastResortStdout(stdout string, limit deliveryLimit) string { minimal := map[string]string{} var existingWarning json.RawMessage if obj, ok := decodeStdoutObject(stdout); ok { @@ -639,8 +704,8 @@ func lastResortStdout(stdout string) string { // maxLastResortFieldBytes: 8 KiB against a ~16 MiB budget is free, and // clipping an operator signal to 1 KiB would defeat the point of keeping it. minimal["warning"] = appendResultWarning(existingWarning, fmt.Sprintf( - "backup result exceeded the %d byte agent IPC limit and was reduced to a terminal status; detail was dropped", - ipc.MaxMessageSize)) + "backup result exceeded the %s and was reduced to a terminal status; detail was dropped", + limit.describe())) data, err := json.Marshal(minimal) if err != nil { return `{"warning":"backup result could not be encoded"}` diff --git a/agent/cmd/breeze-backup/result_bounds_server_cap_test.go b/agent/cmd/breeze-backup/result_bounds_server_cap_test.go index 3960d92b4..c12daa70c 100644 --- a/agent/cmd/breeze-backup/result_bounds_server_cap_test.go +++ b/agent/cmd/breeze-backup/result_bounds_server_cap_test.go @@ -2,6 +2,7 @@ package main import ( "encoding/json" + "fmt" "strings" "testing" @@ -39,7 +40,7 @@ func TestFourThousandFileRunIsDegradedForTheServerCap(t *testing.T) { len(result.Stdout), wire.MaxCommandResultBytes) } - fitted, notes, limitName, limitBytes := fitBackupResult(result) + fitted, notes, limit := fitBackupResult(result) if notes == "" { t.Fatal("a result over the server's `result` cap was passed through undegraded — this is #3001") @@ -48,16 +49,25 @@ func TestFourThousandFileRunIsDegradedForTheServerCap(t *testing.T) { t.Fatalf("degraded stdout is %d bytes, still over the server budget %d", len(fitted.Stdout), serverResultBudget) } - if limitName != limitServerResult { - t.Fatalf("degradation attributed to %q, want the server result cap", limitName) + if limit.name != limitServerResult.name { + t.Fatalf("degradation attributed to %q, want the server result cap", limit.name) } - if limitBytes != wire.MaxCommandResultBytes { - t.Fatalf("reported limitBytes = %d, want %d", limitBytes, wire.MaxCommandResultBytes) + if limit.budget != serverResultBudget { + t.Fatalf("reported budget = %d, want the threshold actually crossed (%d)", + limit.budget, serverResultBudget) + } + if limit.cap != wire.MaxCommandResultBytes { + t.Fatalf("reported cap = %d, want %d", limit.cap, wire.MaxCommandResultBytes) } if !strings.Contains(notes, "snapshot file index dropped") { t.Fatalf("expected the per-file index to be the thing dropped, got notes %q", notes) } + // The warning PERSISTED to backup_jobs.errorLog must name the same limit + // the log line does. Hardcoding the IPC limit here told the customer their + // 2 MB result had overflowed a 16 MiB frame. + assertWarningNamesLimit(t, fitted.Stdout, limit) + assertTerminalStatusSurvives(t, fitted, result.CommandID) } @@ -68,13 +78,13 @@ func TestFourThousandFileRunIsDegradedForTheServerCap(t *testing.T) { func TestTwelveHundredFileRunIsSentIntact(t *testing.T) { result := mustRunResult(t, buildLargeRunJob(qaPassingFileCount, 0)) - fitted, notes, limitName, _ := fitBackupResult(result) + fitted, notes, limit := fitBackupResult(result) if notes != "" { t.Fatalf("a 1,200-file run was degraded (%q); it fits the server cap and must be sent intact", notes) } - if limitName != "" { - t.Fatalf("no limit should have been reported for an in-budget result, got %q", limitName) + if limit.fired() { + t.Fatalf("no limit should have been reported for an in-budget result, got %q", limit.name) } if fitted.Stdout != result.Stdout { t.Fatal("stdout was modified for an in-budget result") @@ -97,11 +107,12 @@ func TestTwelveHundredFileRunIsSentIntact(t *testing.T) { func TestHundredThousandFileRunStillReportsCompletion(t *testing.T) { result := mustRunResult(t, buildLargeRunJob(100000, 0)) - fitted, notes, _, _ := fitBackupResult(result) + fitted, notes, limit := fitBackupResult(result) if notes == "" { t.Fatal("a 100k-file result was not degraded at all") } + assertWarningNamesLimit(t, fitted.Stdout, limit) if len(fitted.Stdout) > serverResultBudget { t.Fatalf("degraded stdout is %d bytes, over the server budget %d — the server would refuse it "+ "and the job would be reaped as stalled", len(fitted.Stdout), serverResultBudget) @@ -142,18 +153,18 @@ func TestStderrOnlyDegradationNamesTheTextCap(t *testing.T) { len(result.Stderr), maxResultTextBytes) } - fitted, notes, limitName, limitBytes := fitBackupResult(result) + fitted, notes, limit := fitBackupResult(result) if notes == "" { t.Fatal("oversize stderr was not truncated") } - if limitName != limitResultText { - t.Fatalf("attributed the degradation to %q; the free-text cap is what fired", limitName) + if limit.name != limitResultText.name { + t.Fatalf("attributed the degradation to %q; the free-text cap is what fired", limit.name) } - if limitBytes != maxResultTextBytes { - t.Fatalf("reported limitBytes = %d, want the text cap %d", limitBytes, maxResultTextBytes) + if limit.budget != maxResultTextBytes { + t.Fatalf("reported budget = %d, want the text cap %d", limit.budget, maxResultTextBytes) } - if limitBytes == ipc.MaxMessageSize { + if limit.cap == ipc.MaxMessageSize { t.Fatal("still reporting the IPC frame size for a degradation the IPC frame did not cause") } if marshalledSize(fitted) > serverResultBudget { @@ -168,7 +179,7 @@ func TestDeliveryWrapperMatchesAttributedForm(t *testing.T) { result := mustRunResult(t, buildLargeRunJob(qaResidualFileCount, 3)) wrappedResult, wrappedNotes := fitBackupResultForDelivery(result) - fullResult, fullNotes, _, _ := fitBackupResult(result) + fullResult, fullNotes, _ := fitBackupResult(result) if wrappedNotes != fullNotes { t.Fatalf("wrapper notes %q != %q", wrappedNotes, fullNotes) @@ -178,6 +189,150 @@ func TestDeliveryWrapperMatchesAttributedForm(t *testing.T) { } } +// TestPersistedWarningNamesTheLimitThatFired is the operator-facing half of +// requirement 3, and the half that reaches the customer. +// +// The structured log line stays on the endpoint; the `warning` these tiers +// write is persisted to backup_jobs.errorLog and rendered in the UI. All five +// tier warnings used to hardcode ipc.MaxMessageSize, so after this PR shifted +// the dominant trigger to the 1 MiB server cap, the headline repro would have +// told a customer that a 2 MB result exceeded a 16 MiB limit — false on its +// face, and self-contradictory in oversizeFailureResult, which prints the +// actual size right next to the limit it supposedly exceeded. +func TestPersistedWarningNamesTheLimitThatFired(t *testing.T) { + t.Run("snapshot index dropped for the server cap", func(t *testing.T) { + result := mustRunResult(t, buildLargeRunJob(qaResidualFileCount, 3)) + fitted, _, limit := fitBackupResult(result) + warning := warningFromStdout(t, fitted.Stdout) + + if !strings.Contains(warning, "server result budget") { + t.Fatalf("warning does not name the server result budget: %q", warning) + } + if strings.Contains(warning, "agent IPC") { + t.Fatalf("warning still blames the agent IPC limit for a server-cap degradation: %q", warning) + } + if !strings.Contains(warning, fmt.Sprint(limit.budget)) { + t.Fatalf("warning does not carry the budget %d that was actually crossed: %q", limit.budget, warning) + } + if strings.Contains(warning, fmt.Sprint(ipc.MaxMessageSize)) { + t.Fatalf("warning still contains the IPC frame size %d: %q", ipc.MaxMessageSize, warning) + } + }) + + t.Run("non-object body degraded to an oversize failure", func(t *testing.T) { + // backup_list's array body: it cannot be summarised, so tier 2 replaces + // it with an explicit failure. This is the site where the wrong limit + // was most visibly self-contradictory. + big := make([]string, 0, 40000) + for i := 0; i < 40000; i++ { + big = append(big, strings.Repeat("s", 40)) + } + encoded, err := json.Marshal(big) + if err != nil { + t.Fatalf("marshal fixture: %v", err) + } + result := backupipc.BackupCommandResult{CommandID: "c1", Success: true, Stdout: string(encoded)} + + fitted, notes, limit := fitBackupResult(result) + if notes == "" { + t.Fatal("an oversize array body was not degraded") + } + if fitted.Success { + t.Fatal("an unsummarisable oversize body must degrade to an explicit failure, not an empty success") + } + if !strings.Contains(fitted.Stderr, limit.describe()) { + t.Fatalf("failure text does not name the limit that fired (%s): %q", limit.describe(), fitted.Stderr) + } + if strings.Contains(fitted.Stderr, fmt.Sprint(ipc.MaxMessageSize)) { + t.Fatalf("failure text still names the IPC frame size: %q", fitted.Stderr) + } + }) +} + +// TestIPCFrameAttributionForOversizeStderr covers the OTHER delivery limit. +// +// Stdout stays under the server budget while a colossal stderr pushes the raw +// sum past the IPC budget, so the IPC frame is genuinely the binding limit — +// the one case where naming it is correct. Without this, every attribution +// assertion in this file could pass with the server cap hardcoded, which is the +// same mistake in the opposite direction. +func TestIPCFrameAttributionForOversizeStderr(t *testing.T) { + result := backupipc.BackupCommandResult{ + CommandID: "c1", + Success: false, + Stdout: `{"id":"job-1","status":"failed"}`, + Stderr: strings.Repeat("x", resultPayloadBudget+1), + } + if len(result.Stdout) > serverResultBudget { + t.Fatal("fixture stdout must stay UNDER the server budget so the IPC frame is the binding limit") + } + + limit := exceededLimit(result) + if limit.name != limitIPCFrame.name { + t.Fatalf("attributed to %q, want the IPC frame — stdout is in budget and only the raw sum is over", + limit.name) + } + if limit.budget != resultPayloadBudget { + t.Fatalf("reported budget = %d, want the IPC budget %d", limit.budget, resultPayloadBudget) + } + if limit.cap != ipc.MaxMessageSize { + t.Fatalf("reported cap = %d, want the IPC frame size %d", limit.cap, ipc.MaxMessageSize) + } +} + +// TestDeliveryLimitReportsTheThresholdActuallyCrossed guards the budget/cap +// split. Reporting the cap as the thing "exceeded" would describe a payload of +// 1,000,000 bytes — over the 983,040 budget, under the 1,048,576 cap — as +// having overflowed a limit it never reached. +func TestDeliveryLimitReportsTheThresholdActuallyCrossed(t *testing.T) { + between := serverResultBudget + (wire.MaxCommandResultBytes-serverResultBudget)/2 + result := backupipc.BackupCommandResult{ + CommandID: "c1", + Success: true, + Stdout: `{"pad":"` + strings.Repeat("x", between) + `"}`, + } + + limit := exceededLimit(result) + if limit.name != limitServerResult.name { + t.Fatalf("a body between the budget and the cap must trip the server limit, got %q", limit.name) + } + if limit.budget >= limit.cap { + t.Fatalf("budget (%d) must be strictly below cap (%d) for the server limit", limit.budget, limit.cap) + } + if !strings.Contains(limit.describe(), fmt.Sprint(limit.budget)) { + t.Fatalf("describe() must state the budget that was crossed, got %q", limit.describe()) + } + if strings.Contains(limit.describe(), fmt.Sprint(limit.cap)) { + t.Fatalf("describe() must not present the cap as the threshold exceeded, got %q", limit.describe()) + } +} + +// warningFromStdout extracts the `warning` field the tiers write into the run +// body — the string the server persists to backup_jobs.errorLog. +func warningFromStdout(t *testing.T, stdout string) string { + t.Helper() + var body struct { + Warning string `json:"warning"` + } + if err := json.Unmarshal([]byte(stdout), &body); err != nil { + t.Fatalf("unmarshal degraded stdout: %v", err) + } + if body.Warning == "" { + t.Fatal("degraded result carries no warning; the operator has no signal at all") + } + return body.Warning +} + +// assertWarningNamesLimit checks that the persisted warning names the limit the +// structured log names, so the two can never tell an operator different stories. +func assertWarningNamesLimit(t *testing.T, stdout string, limit deliveryLimit) { + t.Helper() + warning := warningFromStdout(t, stdout) + if !strings.Contains(warning, limit.describe()) { + t.Fatalf("persisted warning does not name the limit that fired (%s): %q", limit.describe(), warning) + } +} + // assertTerminalStatusSurvives checks the invariant every tier exists to // protect: whatever else is dropped, the server must still learn that the // command finished and how. diff --git a/agent/cmd/breeze-backup/result_bounds_test.go b/agent/cmd/breeze-backup/result_bounds_test.go index da6399745..a039c67ad 100644 --- a/agent/cmd/breeze-backup/result_bounds_test.go +++ b/agent/cmd/breeze-backup/result_bounds_test.go @@ -482,7 +482,7 @@ func TestFitBackupResultRejectsUnsummarisableBody(t *testing.T) { if fitted.Stdout != "" { t.Errorf("expected an empty stdout, got %.80q", fitted.Stdout) } - if !strings.Contains(fitted.Stderr, "IPC limit") { + if !strings.Contains(fitted.Stderr, limitExceededPhrase) { t.Errorf("expected the stderr to explain the oversize, got %.120q", fitted.Stderr) } } @@ -666,7 +666,7 @@ func TestFitBackupResultLastResortRecoversStatus(t *testing.T) { if out.ID != "job-9" { t.Errorf("expected tier 4 to recover the job id, got %q", out.ID) } - if !strings.Contains(out.Warning, "IPC limit") { + if !strings.Contains(out.Warning, limitExceededPhrase) { t.Errorf("expected tier 4 to explain itself in the warning, got %q", out.Warning) } } diff --git a/agent/cmd/breeze-backup/result_warning_preservation_test.go b/agent/cmd/breeze-backup/result_warning_preservation_test.go index 6ed9699bf..747acd399 100644 --- a/agent/cmd/breeze-backup/result_warning_preservation_test.go +++ b/agent/cmd/breeze-backup/result_warning_preservation_test.go @@ -254,7 +254,7 @@ func TestTier4PreservesOperatorWarning(t *testing.T) { } // The tier still has to say what it did — a preserved warning that hid the // truncation would be its own kind of lie. - if !strings.Contains(warning, "IPC limit") { + if !strings.Contains(warning, limitExceededPhrase) { t.Errorf("tier 4 stopped explaining itself; warning = %.300q", warning) } } @@ -272,7 +272,7 @@ func TestTier4BoundsThePreservedWarning(t *testing.T) { if len(warning) > maxResultTextBytes { t.Errorf("tier 4 left the warning unbounded at %d bytes", len(warning)) } - if !strings.Contains(warning, "IPC limit") { + if !strings.Contains(warning, limitExceededPhrase) { t.Errorf("tier 4 stopped explaining itself; warning = %.300q", warning) } } @@ -312,7 +312,7 @@ func TestBoundingNotesAreIndividuallyCapped(t *testing.T) { if !strings.Contains(warning, "read from the live volume") { t.Errorf("the operator signal was evicted by an oversized note; warning = %.300q", warning) } - if !strings.Contains(warning, "IPC limit") { + if !strings.Contains(warning, limitExceededPhrase) { t.Errorf("an oversized earlier note crowded the last tier's note out of the reserve; warning = %.400q", warning) } } diff --git a/agent/internal/websocket/client.go b/agent/internal/websocket/client.go index c3f6d07cd..f9361a3a2 100644 --- a/agent/internal/websocket/client.go +++ b/agent/internal/websocket/client.go @@ -747,18 +747,33 @@ func (c *Client) processCommand(cmd Command) { // re-persists it on failure. func (c *Client) SendResult(result CommandResult) error { data, err := json.Marshal(result) - if err != nil { - return fmt.Errorf("failed to marshal result: %w", err) - } - if len(data) > wire.CommandResultBudget { + switch { + case err != nil: + // The body is the ONLY field that can fail to encode — every other + // field on CommandResult is a string or an int — so an encoding error + // means the body is at fault and dropping it is both sufficient and + // correct. Returning here instead would lose the terminal status to an + // unencodable detail field (a NaN in a metrics map is enough), which is + // precisely the trade #3001 exists to prevent: detail is expendable, + // the terminal status is not. + bounded, dropped := boundResultFieldForServer(result) + if !dropped { + // No body to drop, so the failure is something this cannot repair. + return fmt.Errorf("failed to marshal result: %w", err) + } + result = bounded + if data, err = json.Marshal(result); err != nil { + return fmt.Errorf("failed to marshal result after dropping its body: %w", err) + } + + case len(data) > wire.CommandResultBudget: // Only pay for the precise per-field measurement once the whole frame // is over the budget — the `result` field's encoded bytes are a subset // of the frame's, so a frame under the budget cannot contain a field // over it. - if bounded, ok := boundResultFieldForServer(result); ok { + if bounded, dropped := boundResultFieldForServer(result); dropped { result = bounded - data, err = json.Marshal(result) - if err != nil { + if data, err = json.Marshal(result); err != nil { return fmt.Errorf("failed to marshal bounded result: %w", err) } } diff --git a/agent/internal/websocket/result_server_cap_test.go b/agent/internal/websocket/result_server_cap_test.go index 5418866e8..87ba13ef0 100644 --- a/agent/internal/websocket/result_server_cap_test.go +++ b/agent/internal/websocket/result_server_cap_test.go @@ -2,6 +2,7 @@ package websocket import ( "encoding/json" + "math" "strings" "testing" @@ -181,6 +182,70 @@ func TestBoundResultFieldHandlesUnmarshallableBody(t *testing.T) { } } +// TestSendResultRecoversFromAnUnencodableBody exercises the marshal-error +// recovery through the WIRED path, which is the only path that matters. +// +// The unit test for boundResultFieldForServer's marshal-error branch passed +// while that branch was unreachable in production: SendResult marshalled the +// whole result first and returned on error, so an unencodable body still lost +// its terminal status and the test asserted a repair that never ran. A NaN in +// any nested map is enough to trigger it — encoding/json rejects non-finite +// floats — and metrics-shaped results carry floats routinely. +func TestSendResultRecoversFromAnUnencodableBody(t *testing.T) { + c := &Client{ + resultChan: make(chan outboundResult, 1), + done: make(chan struct{}), + } + + // Sanity: this really is unencodable, so the test cannot pass vacuously. + if _, err := json.Marshal(map[string]any{"cpu": math.NaN()}); err == nil { + t.Fatal("fixture encodes cleanly; it must fail to marshal for this test to mean anything") + } + + if err := c.SendResult(CommandResult{ + Type: "command_result", + CommandID: "cmd-nan", + Status: "completed", + ExitCode: 0, + Result: map[string]any{"samples": []any{map[string]any{"cpu": math.NaN()}}}, + }); err != nil { + t.Fatalf("SendResult returned %v; an unencodable BODY must not cost the terminal status", err) + } + + select { + case queued := <-c.resultChan: + var decoded struct { + CommandID string `json:"commandId"` + Status string `json:"status"` + Result map[string]any `json:"result"` + } + if err := json.Unmarshal(queued.data, &decoded); err != nil { + t.Fatalf("unmarshal enqueued frame: %v", err) + } + if decoded.Status != "completed" || decoded.CommandID != "cmd-nan" { + t.Fatalf("terminal status lost: %+v", decoded) + } + if decoded.Result[resultOmittedMarker] != true { + t.Fatalf("enqueued frame does not carry the omission marker: %v", decoded.Result) + } + default: + t.Fatal("SendResult enqueued nothing; the terminal status was dropped") + } +} + +// TestSendResultStillErrorsWhenThereIsNoBodyToDrop pins the other side of that +// recovery: it must repair an unencodable BODY, not swallow every marshal +// failure. With no body there is nothing to drop and the caller has to hear +// about it. +func TestSendResultStillErrorsWhenThereIsNoBodyToDrop(t *testing.T) { + // Result is nil, so boundResultFieldForServer reports nothing to drop. + // Nothing else on CommandResult can fail to encode, so this is a + // contract test rather than a reachable production path. + if _, dropped := boundResultFieldForServer(CommandResult{CommandID: "c1"}); dropped { + t.Fatal("a nil body must report nothing dropped, or the error path above would swallow real failures") + } +} + // TestSendResultBoundsOversizeBodyBeforeEnqueue proves the backstop is actually // wired into the send path, not merely present. It asserts on the bytes that // reach resultChan, which is what writePump puts on the wire. diff --git a/agent/internal/websocket/server_error_frame_test.go b/agent/internal/websocket/server_error_frame_test.go new file mode 100644 index 000000000..2b5d2c147 --- /dev/null +++ b/agent/internal/websocket/server_error_frame_test.go @@ -0,0 +1,120 @@ +package websocket + +import ( + "encoding/json" + "os" + "path/filepath" + "strings" + "testing" +) + +// serverRejectionSource is the TypeScript that BUILDS the frame this file +// parses. Relative to this package's directory. +const serverRejectionSource = "../../../apps/api/src/routes/agentWs.ts" + +// canonicalRejectionFrame is the frame the server emits for a rejected +// command_result, as pinned on the other side by agentWs.rejectionFrame.test.ts. +const canonicalRejectionFrame = `{ + "type": "error", + "code": "INVALID_MESSAGE", + "message": "Invalid message format", + "messageType": "command_result", + "commandId": "cmd-7", + "details": [{"code":"custom","path":["result"],"message":"Command result payload exceeds the 1048576-byte ` + "`result`" + ` limit"}] +}` + +// TestServerErrorFrameParsesEveryAttributionField is the agent half of the +// error-frame contract. +// +// #3001's defining symptom was that the agent had NO trace of a rejected +// terminal result: the write succeeded, so every send path reported success, +// and the server's explanation was discarded on arrival. These four fields are +// the entire remedy, so each is asserted individually — a frame that parses but +// yields an empty commandId is worth almost nothing to an operator trying to +// find which job died. +func TestServerErrorFrameParsesEveryAttributionField(t *testing.T) { + var frame struct { + Code string `json:"code"` + Message string `json:"message"` + MessageType string `json:"messageType"` + CommandID string `json:"commandId"` + Details json.RawMessage `json:"details"` + } + if err := json.Unmarshal([]byte(canonicalRejectionFrame), &frame); err != nil { + t.Fatalf("the canonical server rejection frame does not parse: %v", err) + } + + for _, tc := range []struct{ name, got, want string }{ + {"code", frame.Code, "INVALID_MESSAGE"}, + {"messageType", frame.MessageType, "command_result"}, + {"commandId", frame.CommandID, "cmd-7"}, + {"message", frame.Message, "Invalid message format"}, + } { + if tc.got != tc.want { + t.Errorf("%s = %q, want %q — logServerErrorFrame would log this rejection unattributed", + tc.name, tc.got, tc.want) + } + } + if len(frame.Details) == 0 { + t.Error("details did not parse; the operator loses the reason the frame was rejected") + } + + // Does not panic and does not depend on any field being present. + logServerErrorFrame([]byte(canonicalRejectionFrame)) + logServerErrorFrame([]byte(`{"type":"error"}`)) + logServerErrorFrame([]byte(`not json at all`)) +} + +// TestServerErrorFrameFieldNamesMatchTheServer pins the field names against the +// TypeScript that emits them, from this side. The Vitest twin +// (agentWs.rejectionFrame.test.ts) pins the same contract in the other +// direction; either alone can be satisfied by renaming both the emitter and its +// own test. +func TestServerErrorFrameFieldNamesMatchTheServer(t *testing.T) { + path := filepath.Clean(serverRejectionSource) + data, err := os.ReadFile(path) + if err != nil { + t.Fatalf("cannot read the server rejection builder at %s: %v", path, err) + } + source := string(data) + + // The keys logServerErrorFrame depends on to attribute a rejection. + for _, field := range []string{"messageType", "commandId", "code", "details"} { + if !strings.Contains(source, field) { + t.Errorf("the server no longer emits %q in its rejection frame; logServerErrorFrame "+ + "would silently log an empty value for it (issue #3001)", field) + } + } + if !strings.Contains(source, "buildAgentMessageRejection") { + t.Error("buildAgentMessageRejection is gone from agentWs.ts; the frame contract this test " + + "guards has moved and this test must be repointed") + } +} + +// TestReadPumpHandlesErrorFramesBeforeTheIDLessSkip pins the ORDERING that made +// the fix work. +// +// Server rejections carry no `id`, so before this change they fell into the +// "not a command" skip a few lines below and were discarded without a word. +// Moving the `error` branch back under that skip would restore the silence +// while every other test kept passing, so the order is asserted directly. +func TestReadPumpHandlesErrorFramesBeforeTheIDLessSkip(t *testing.T) { + source, err := os.ReadFile("client.go") + if err != nil { + t.Fatalf("read client.go: %v", err) + } + body := string(source) + + errorBranch := strings.Index(body, `if msg.Type == "error" {`) + if errorBranch < 0 { + t.Fatal(`readPump no longer has an "error" branch; server rejections are being discarded again`) + } + idLessSkip := strings.Index(body, `if msg.ID == "" {`) + if idLessSkip < 0 { + t.Fatal("the id-less skip is gone; this test needs repointing") + } + if errorBranch > idLessSkip { + t.Fatal(`the "error" branch now sits AFTER the id-less skip, so error frames (which carry no id) ` + + "are swallowed before it runs — exactly the #3001 silence") + } +} diff --git a/agent/internal/wire/limits_test.go b/agent/internal/wire/limits_test.go index 960f90a5c..4b6a14afb 100644 --- a/agent/internal/wire/limits_test.go +++ b/agent/internal/wire/limits_test.go @@ -38,10 +38,14 @@ func TestMaxCommandResultBytesMatchesServerSchema(t *testing.T) { t.Fatalf("cannot read the server schema at %s to verify the mirrored cap: %v", path, err) } - re := regexp.MustCompile(`MAX_COMMAND_RESULT_BYTES\s*=\s*([0-9_]+)`) + // Anchored on the declaration keyword so a doc comment that happens to + // contain "MAX_COMMAND_RESULT_BYTES = " cannot retarget the pin onto + // prose — which would let the real constant drift while this test kept + // passing against a sentence. + re := regexp.MustCompile(`export\s+const\s+MAX_COMMAND_RESULT_BYTES\s*=\s*([0-9_]+)`) m := re.FindSubmatch(data) if m == nil { - t.Fatalf("no `MAX_COMMAND_RESULT_BYTES = ` declaration found in %s — if it was "+ + t.Fatalf("no `export const MAX_COMMAND_RESULT_BYTES = ` declaration found in %s — if it was "+ "renamed or moved, update serverSchemaPath and this pattern", path) } serverValue, err := strconv.Atoi(strings.ReplaceAll(string(m[1]), "_", "")) diff --git a/apps/api/src/routes/agentWs.rejectionFrame.test.ts b/apps/api/src/routes/agentWs.rejectionFrame.test.ts new file mode 100644 index 000000000..d6d674cf0 --- /dev/null +++ b/apps/api/src/routes/agentWs.rejectionFrame.test.ts @@ -0,0 +1,142 @@ +import { describe, it, expect } from 'vitest'; +import { readFileSync } from 'node:fs'; +import { resolve } from 'node:path'; +import { z } from 'zod'; +import { + buildAgentMessageRejection, + MAX_ECHOED_FIELD_CHARS, + MAX_ECHOED_ISSUES, + MAX_PRECISE_RESULT_MEASURE_BYTES, +} from './agentWs'; +import { MAX_COMMAND_RESULT_BYTES } from './agents/schemas'; + +// The Go parser that consumes the frame this module emits. +const GO_CLIENT_PATH = resolve(__dirname, '../../../../agent/internal/websocket/client.go'); + +function issues(count = 1): z.ZodIssue[] { + return Array.from({ length: count }, (_, i) => ({ + code: 'custom', + path: ['result'], + message: `issue ${i}`, + })) as unknown as z.ZodIssue[]; +} + +describe('agent rejection frame (#3001)', () => { + it('escalates a rejected command_result to error and names the job', () => { + const { level, log, frame } = buildAgentMessageRejection({ + agentId: 'agent-1', + message: { type: 'command_result', commandId: 'cmd-7', status: 'completed', result: { a: 1 } }, + frameBytes: 2_100_000, + issues: issues(), + }); + + // A lost terminal status is not a warning: nothing downstream of this + // branch runs, so this line is the only record the job ever produces. + expect(level).toBe('error'); + expect(log).toContain('cmd-7'); + expect(log).toContain('frameBytes=2100000'); + expect(log).toContain(`resultLimitBytes=${MAX_COMMAND_RESULT_BYTES}`); + expect(log).toContain('reaper'); + expect(frame.commandId).toBe('cmd-7'); + expect(frame.messageType).toBe('command_result'); + }); + + it('keeps other invalid messages at warn', () => { + const { level, log } = buildAgentMessageRejection({ + agentId: 'agent-1', + message: { type: 'heartbeat' }, + frameBytes: 120, + issues: issues(), + }); + expect(level).toBe('warn'); + expect(log).toContain('type=heartbeat'); + }); + + it('skips the precise measurement on a huge frame rather than stalling the event loop', () => { + // No maxPayload is set on the agent WS server, so `ws` accepts up to + // 100 MiB and pre-fix agents still send unbounded results. Re-serialising + // one synchronously to fill a log field is the stall this guard prevents. + const { log } = buildAgentMessageRejection({ + agentId: 'agent-1', + message: { type: 'command_result', commandId: 'c1', result: { a: 1 } }, + frameBytes: MAX_PRECISE_RESULT_MEASURE_BYTES, + issues: issues(), + }); + expect(log).toContain('resultBytes=unmeasured'); + expect(log).toContain(String(MAX_PRECISE_RESULT_MEASURE_BYTES)); + }); + + it('measures precisely just below the threshold', () => { + const { log } = buildAgentMessageRejection({ + agentId: 'agent-1', + message: { type: 'command_result', commandId: 'c1', result: { a: 1 } }, + frameBytes: MAX_PRECISE_RESULT_MEASURE_BYTES - 1, + issues: issues(), + }); + expect(log).toContain(`resultBytes=${JSON.stringify({ a: 1 }).length}`); + }); + + it('reports an unencodable result body without throwing', () => { + const cyclic: Record = {}; + cyclic.self = cyclic; + const { log } = buildAgentMessageRejection({ + agentId: 'agent-1', + message: { type: 'command_result', commandId: 'c1', result: cyclic }, + frameBytes: 500, + issues: issues(), + }); + expect(log).toContain('resultBytes=unencodable'); + }); + + it('clamps agent-supplied strings echoed back in the frame', () => { + // messageType and commandId come off an UNVALIDATED message — the entire + // reason this branch exists — so their length is the agent's choice. + const { frame } = buildAgentMessageRejection({ + agentId: 'agent-1', + message: { type: 'x'.repeat(5000), commandId: 'y'.repeat(5000) }, + frameBytes: 10_000, + issues: issues(50), + }); + expect(frame.messageType.length).toBe(MAX_ECHOED_FIELD_CHARS); + expect(frame.commandId!.length).toBe(MAX_ECHOED_FIELD_CHARS); + expect(frame.details.length).toBe(MAX_ECHOED_ISSUES); + }); + + it('survives a message that is not an object at all', () => { + for (const message of [null, undefined, 5, 'a string', []]) { + const { level, frame } = buildAgentMessageRejection({ + agentId: 'agent-1', + message, + frameBytes: 4, + issues: issues(), + }); + expect(level).toBe('warn'); + expect(frame.messageType).toBe('unknown'); + expect(frame.commandId).toBeUndefined(); + } + }); + + it('emits the exact field names the Go agent parses off the frame', () => { + // THE contract test. logServerErrorFrame in the Go client reads these keys + // to attribute a rejection to a command. A rename on either side puts the + // agent back to logging nothing for a lost terminal status — #3001's + // defining symptom — and nothing else would catch it, because both sides + // keep compiling and every other test keeps passing. + const { frame } = buildAgentMessageRejection({ + agentId: 'agent-1', + message: { type: 'command_result', commandId: 'cmd-7' }, + frameBytes: 100, + issues: issues(), + }); + + const source = readFileSync(GO_CLIENT_PATH, 'utf8'); + for (const [key, value] of Object.entries(frame)) { + expect(value).toBeDefined(); + expect( + source.includes(`json:"${key}"`), + `the Go error-frame parser has no field tagged json:"${key}"; ` + + `agent/internal/websocket/client.go logServerErrorFrame must be updated in the same commit` + ).toBe(true); + } + }); +}); diff --git a/apps/api/src/routes/agentWs.ts b/apps/api/src/routes/agentWs.ts index 135397375..c4a27a3cb 100644 --- a/apps/api/src/routes/agentWs.ts +++ b/apps/api/src/routes/agentWs.ts @@ -330,6 +330,120 @@ const commandResultSchema = baseCommandResultSchema.extend({ type AgentCommandResult = z.infer; +/** + * MAX_PRECISE_RESULT_MEASURE_BYTES bounds when the rejection path is willing to + * re-serialise a `result` body to report its exact size. + * + * `commandResultResultByteLength` runs a synchronous `JSON.stringify` over a + * just-parsed object, and this branch is reached by definition on messages that + * failed validation — including unbounded ones. No `maxPayload` is configured + * on the agent WebSocket server, so `ws` allows frames up to its 100 MiB + * default, and pre-fix agents still in the field send genuinely unbounded + * results (#3001's original report was a 64 MB payload). Stringifying that on + * the event loop to produce a log field is a self-inflicted stall, and this + * repo has form for exactly that (#3236). + * + * Above the threshold the frame size is reported instead — a lower bound that + * is already enough to diagnose an oversize rejection, and free to compute. + */ +export const MAX_PRECISE_RESULT_MEASURE_BYTES = 8_000_000; + +/** + * MAX_ECHOED_FIELD_CHARS clamps the agent-supplied strings echoed back in a + * rejection. They come off an UNVALIDATED message — that is the whole point of + * this branch — so their length is whatever the agent chose to send. + */ +export const MAX_ECHOED_FIELD_CHARS = 200; + +/** MAX_ECHOED_ISSUES clamps how many Zod issues ride back in the reply. */ +export const MAX_ECHOED_ISSUES = 10; + +/** + * buildAgentMessageRejection assembles the log line and the error frame for a + * message that failed schema validation. + * + * #3001: this branch used to be a bare `console.warn` carrying only the raw Zod + * issues. When it rejected a `command_result` — the terminal status of a job + * the server is actively waiting on — the operator saw no commandId, no size + * and no message type, and none of the downstream backup logs, which live in + * `processCommandResult` past this early return. The result read as having + * vanished in transit, and a backup that had succeeded was reaped as stalled 15 + * minutes later. + * + * A rejected `command_result` is a LOST TERMINAL STATUS, so it logs at error + * with everything needed to identify the job; anything else stays a warning. + * + * Pure and exported so the frame shape can be pinned by a test: the agent side + * parses `messageType`/`commandId` off this object (see `logServerErrorFrame` + * in agent/internal/websocket/client.go), and a rename on either side silently + * returns the agent to the no-trace state this issue was about. + */ +export function buildAgentMessageRejection(args: { + agentId: string; + message: unknown; + frameBytes: number; + issues: z.ZodIssue[]; +}): { + level: 'error' | 'warn'; + log: string; + frame: { + type: 'error'; + code: 'INVALID_MESSAGE'; + message: string; + messageType: string; + commandId?: string; + details: z.ZodIssue[]; + }; +} { + const { agentId, message, frameBytes, issues } = args; + // `message` is unvalidated and need not even be an object (a bare JSON number + // or null both reach here), so every read is guarded. + const raw = (message ?? {}) as Record; + const clamp = (v: unknown): string | undefined => + typeof v === 'string' ? v.slice(0, MAX_ECHOED_FIELD_CHARS) : undefined; + + const messageType = clamp(raw.type) ?? 'unknown'; + const commandId = clamp(raw.commandId); + const details = issues.slice(0, MAX_ECHOED_ISSUES); + const frame = { + type: 'error' as const, + code: 'INVALID_MESSAGE' as const, + message: 'Invalid message format', + // Echoed so the agent can attribute the rejection to the command it sent. + // Without these the agent sees an unattributable error frame and has + // nothing to log against the job. + messageType, + ...(commandId !== undefined ? { commandId } : {}), + details, + }; + + if (messageType !== 'command_result') { + return { + level: 'warn', + log: `Invalid message from agent ${agentId} (type=${messageType}, frameBytes=${frameBytes}):`, + frame, + }; + } + + let resultBytes: string; + if (frameBytes >= MAX_PRECISE_RESULT_MEASURE_BYTES) { + resultBytes = `unmeasured(frame ${frameBytes}B exceeds the ${MAX_PRECISE_RESULT_MEASURE_BYTES}B measure threshold)`; + } else { + const measured = commandResultResultByteLength(raw.result); + resultBytes = measured === null ? 'unencodable' : String(measured); + } + + return { + level: 'error', + log: + `[AgentWs] REJECTED command_result from agent ${agentId} — the job will have no ` + + `terminal status and will be failed by a reaper. commandId=${commandId ?? 'unknown'} ` + + `frameBytes=${frameBytes} resultBytes=${resultBytes} ` + + `resultLimitBytes=${MAX_COMMAND_RESULT_BYTES}:`, + frame, + }; +} + function commandResultToStdout(result: AgentCommandResult): string | undefined { return result.stdout ?? (result.result !== undefined ? JSON.stringify(result.result) : undefined); @@ -2289,47 +2403,18 @@ export function createAgentWsHandlers(agentId: string, preValidatedAgent: AgentD const parsed = agentMessageSchema.safeParse(message); if (!parsed.success) { - // #3001: this branch used to be a bare console.warn carrying only the - // raw Zod issues. When it rejected a `command_result` — the terminal - // status of a job the server is actively waiting on — the operator - // saw no commandId, no size, no message type, and none of the - // downstream backup logs (those live in processCommandResult, past - // this return). The result read as having vanished in transit, and a - // succeeded backup was reaped as stalled 15 minutes later. - // - // A rejected command_result is a LOST TERMINAL STATUS, so it logs at - // error with everything needed to identify the job and the offending - // field; anything else stays a warning. - const rejectedType = typeof message?.type === 'string' ? message.type : 'unknown'; - const rejectedCommandId = typeof message?.commandId === 'string' - ? message.commandId - : undefined; - if (rejectedType === 'command_result') { - const resultBytes = commandResultResultByteLength(message?.result); - console.error( - `[AgentWs] REJECTED command_result from agent ${agentId} — the job will have no ` + - `terminal status and will be failed by a reaper. commandId=${rejectedCommandId ?? 'unknown'} ` + - `frameBytes=${data.length} resultBytes=${resultBytes ?? 'unencodable'} ` + - `resultLimitBytes=${MAX_COMMAND_RESULT_BYTES}:`, - parsed.error.issues - ); + const rejection = buildAgentMessageRejection({ + agentId, + message, + frameBytes: Buffer.byteLength(data, 'utf8'), + issues: parsed.error.issues, + }); + if (rejection.level === 'error') { + console.error(rejection.log, rejection.frame.details); } else { - console.warn( - `Invalid message from agent ${agentId} (type=${rejectedType}, frameBytes=${data.length}):`, - parsed.error.issues - ); + console.warn(rejection.log, rejection.frame.details); } - ws.send(JSON.stringify({ - type: 'error', - code: 'INVALID_MESSAGE', - message: 'Invalid message format', - // Echoed so the agent can attribute the rejection to the command it - // sent. Without it the agent sees an unattributable error frame and - // has nothing to log against the job. - messageType: rejectedType, - commandId: rejectedCommandId, - details: parsed.error.issues - })); + ws.send(JSON.stringify(rejection.frame)); return; } diff --git a/apps/api/src/routes/agents/schemas.commandResult.test.ts b/apps/api/src/routes/agents/schemas.commandResult.test.ts index ed5d80cfd..5220e79f4 100644 --- a/apps/api/src/routes/agents/schemas.commandResult.test.ts +++ b/apps/api/src/routes/agents/schemas.commandResult.test.ts @@ -41,9 +41,12 @@ describe('command_result `result` size cap (#3001)', () => { it('matches the Go mirror in agent/internal/wire/limits.go', () => { const source = readFileSync(GO_LIMITS_PATH, 'utf8'); - const declaration = source.match(/MaxCommandResultBytes\s*=\s*([0-9*\s]+)/)?.[1]; + // Anchored on `const` so a doc comment containing + // "MaxCommandResultBytes = " cannot retarget the pin onto prose, + // letting the real constant drift while this test passes against a sentence. + const declaration = source.match(/const\s+MaxCommandResultBytes\s*=\s*([0-9*\s]+)/)?.[1]; if (declaration === undefined) { - throw new Error(`no MaxCommandResultBytes declaration found in ${GO_LIMITS_PATH}`); + throw new Error(`no \`const MaxCommandResultBytes = \` declaration found in ${GO_LIMITS_PATH}`); } // The Go side writes it as an arithmetic literal (1024 * 1024).