From 06f0317bc6886d42be88648c89c4ae650dce1eef Mon Sep 17 00:00:00 2001 From: santoshkumarradha Date: Thu, 17 Sep 2026 22:07:05 -0400 Subject: [PATCH 1/4] outbox: a test for a refusal that names its line Co-Authored-By: codeaf --- internal/pool/outbox/outbox_test.go | 95 +++++++++++++++++++++++++++++ 1 file changed, 95 insertions(+) diff --git a/internal/pool/outbox/outbox_test.go b/internal/pool/outbox/outbox_test.go index 36b81e524f..e217ffebd8 100644 --- a/internal/pool/outbox/outbox_test.go +++ b/internal/pool/outbox/outbox_test.go @@ -696,6 +696,101 @@ func TestSendSaysWhatFailedAndLeavesTheRowsPending(t *testing.T) { } } +func TestARefusedRowLeavesTheOutboxAndTheRestOfItsBatchSends(t *testing.T) { + // The relay refuses a batch on the first line it cannot validate and + // answers which line that was, so the client can retire exactly that row + // and send the rows around it rather than hold every row behind it + // pending on every run to come. + var mu sync.Mutex + var bodies []string + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + b, _ := io.ReadAll(r.Body) + mu.Lock() + bodies = append(bodies, string(b)) + lines := strings.Count(string(b), "\n") + mu.Unlock() + if lines == 3 { + w.WriteHeader(400) + w.Write([]byte(`{"error":"line 2: role must be worker, high or mastermind"}`)) + return + } + w.WriteHeader(202) + })) + defer srv.Close() + o, path := openOutbox(t) + o.Rand = &stepRand{step: 14} + appendRow(t, o, `{"k":"worker"}`) + appendRow(t, o, `{"role":"scout"}`) + appendRow(t, o, `{"k":"mastermind"}`) + bad := pendingNonces(t, o)[1] + n, err := o.Send(context.Background(), srv.URL) + if n != 2 || err != nil { + t.Fatalf("Send answered (%d, %v), want (2, nil)", n, err) + } + mu.Lock() + defer mu.Unlock() + if len(bodies) != 2 { + t.Fatalf("the send made %d posts, want one per refused line", len(bodies)) + } + if strings.Contains(bodies[1], bad) { + t.Fatal("the retried batch carried the refused row again") + } + if cn := countMarks(t, path, "sent"); cn != 2 { + t.Fatalf("file holds %d sent markers, want 2", cn) + } + if cn := countMarks(t, path, "dropped"); cn != 1 { + t.Fatalf("file holds %d dropped markers, want 1", cn) + } + want := `{"dropped":"` + bad + `"}` + found := false + for _, line := range fileLines(t, path) { + if line == want { + found = true + } + } + if !found { + t.Fatalf("no dropped marker names the refused row, want %s", want) + } + if got := o.Pending(); len(got) != 0 { + t.Fatalf("%d rows stayed pending after the refused row was retired, want 0", len(got)) + } + if err := o.Close(); err != nil { + t.Fatalf("Close: %v", err) + } + o2, err := Open(path) + if err != nil { + t.Fatalf("reopen: %v", err) + } + defer o2.Close() + if got := pendingNonces(t, o2); len(got) != 0 { + t.Fatalf("the dropped mark did not survive a reopen: %d rows pending", len(got)) + } +} + +func TestAQuotaRefusalKeepsEveryRowPending(t *testing.T) { + rec := &postRecord{status: []int{429}} + srv := httptest.NewServer(http.HandlerFunc(rec.handle)) + defer srv.Close() + o, path := openOutbox(t) + o.Rand = &stepRand{step: 15} + for i := 0; i < 3; i++ { + appendRow(t, o, fmt.Sprintf(`{"k":%d}`, i)) + } + n, err := o.Send(context.Background(), srv.URL) + if n != 0 || err == nil { + t.Fatalf("Send answered (%d, %v), want (0, an error)", n, err) + } + if !strings.Contains(err.Error(), "429") { + t.Fatalf("the error does not say what failed: %v", err) + } + if got := len(o.Pending()); got != 3 { + t.Fatalf("%d rows stayed pending after the quota refusal, want 3", got) + } + if cn := countMarks(t, path, "sent") + countMarks(t, path, "dropped"); cn != 0 { + t.Fatalf("file holds %d markers after a quota refusal, want none", cn) + } +} + func TestSendAnswersARefusedConnection(t *testing.T) { o, _ := openOutbox(t) o.Rand = &stepRand{step: 12} From 062e16d3cc8dea2f0b59de06daf1297b0a7d3b02 Mon Sep 17 00:00:00 2001 From: santoshkumarradha Date: Thu, 17 Sep 2026 22:07:05 -0400 Subject: [PATCH 2/4] outbox: a refused row leaves the outbox and the rest of its batch is sent Co-Authored-By: codeaf --- internal/pool/outbox/outbox.go | 99 +++++++++++++++++++++++++++------- 1 file changed, 81 insertions(+), 18 deletions(-) diff --git a/internal/pool/outbox/outbox.go b/internal/pool/outbox/outbox.go index 919b05191a..05c292ebf0 100644 --- a/internal/pool/outbox/outbox.go +++ b/internal/pool/outbox/outbox.go @@ -25,6 +25,7 @@ import ( "net/http" "os" "path/filepath" + "strconv" "strings" "sync" "time" @@ -199,9 +200,12 @@ func (o *Outbox) Pending() []Row { // The whole call lives inside Budget, which is 2s when it is zero, and inside // ctx as well, whichever ends first. Rows whose batch arrived are marked sent // and leave the outbox; a row Keep declines is marked sent without being -// transmitted and is not counted. Rows whose batch did not arrive, and every -// row after it, stay pending, and Send returns an error that says what -// failed. +// transmitted and is not counted. A batch the destination refused by line — +// a 400 naming `line N` — is refused on one row it has said it will never +// accept: the named row is marked dropped, the rows around it are sent +// within the same call, and Send counts the sent ones. Rows whose batch did +// not arrive, and every row after it, stay pending, and Send returns an +// error that says what failed. func (o *Outbox) Send(ctx context.Context, dest string) (int, error) { if dest == "" { return 0, nil @@ -422,15 +426,22 @@ func (o *Outbox) load() []Row { } // sendHTTP posts the rows to dest, at most maxBatch per POST, and marks each -// batch that came back 2xx. The first batch that fails ends the call; its -// rows and every row after it stay pending. +// batch that came back 2xx. A batch refused by line — a 400 whose reply names +// `line N` — is refused on one row the destination will never take: that row +// is marked dropped, so it stays gone across a reopen, and the rest of the +// batch is posted again in the same call. Each retry retires a row, so a +// batch of refused rows costs one POST per row at the most and never a spin. +// The first batch that fails any other way ends the call, its rows and every +// row after it staying pending: a 400 that names no line and a 413 name no +// row at all, so retiring one would be a guess, and 429 and 5xx are about +// the destination rather than any row. func (o *Outbox) sendHTTP(ctx context.Context, dest string, rows []Row) (int, error) { client := o.Client if client == nil { client = defaultClient } n := 0 - for start := 0; start < len(rows); start += maxBatch { + for start := 0; start < len(rows); { end := start + maxBatch if end > len(rows) { end = len(rows) @@ -438,17 +449,36 @@ func (o *Outbox) sendHTTP(ctx context.Context, dest string, rows []Row) (int, er if err := ctx.Err(); err != nil { return n, err } - body := make([]byte, 0, (end-start)*128) - for _, r := range rows[start:end] { + sent, err := o.postBatch(ctx, client, dest, rows[start:end]) + if err != nil { + return n, err + } + if len(sent) > 0 { + o.retire(sentMarks(rowNonces(sent))) + n += len(sent) + } + start = end + } + return n, nil +} + +// postBatch posts one batch and answers the rows of it that arrived. A reply +// that names a line takes that row out and the batch is posted again; a batch +// every row of which was refused answers no rows and no error, the way a +// Keep that declines everything does. +func (o *Outbox) postBatch(ctx context.Context, client *http.Client, dest string, batch []Row) ([]Row, error) { + for len(batch) > 0 { + body := make([]byte, 0, len(batch)*128) + for _, r := range batch { line, err := rowLine(r) if err != nil { - return n, err + return nil, err } body = append(body, line...) } req, err := http.NewRequestWithContext(ctx, http.MethodPost, dest, bytes.NewReader(body)) if err != nil { - return n, err + return nil, err } req.Header.Set("Content-Type", "application/x-ndjson") if o.Install != "" { @@ -456,18 +486,51 @@ func (o *Outbox) sendHTTP(ctx context.Context, dest string, rows []Row) (int, er } resp, err := client.Do(req) if err != nil { - return n, err + return nil, err } - ok := resp.StatusCode >= 200 && resp.StatusCode < 300 - io.CopyN(io.Discard, resp.Body, 1<<16) + reply, _ := io.ReadAll(io.LimitReader(resp.Body, 1<<16)) resp.Body.Close() - if !ok { - return n, fmt.Errorf("outbox: %s replied %s", dest, resp.Status) + if resp.StatusCode >= 200 && resp.StatusCode < 300 { + return batch, nil } - o.retire(sentMarks(rowNonces(rows[start:end]))) - n += end - start + if line := refusedLine(reply, batch); line >= 0 { + o.retire(droppedMarks([]string{batch[line].Nonce})) + rest := make([]Row, 0, len(batch)-1) + rest = append(rest, batch[:line]...) + rest = append(rest, batch[line+1:]...) + batch = rest + continue + } + return nil, fmt.Errorf("outbox: %s replied %s", dest, resp.Status) } - return n, nil + return nil, nil +} + +// refusedLine reads a refusal reply and answers which row of the batch it +// refuses, or -1 when it names no line of the batch. The relay refuses the +// first row that fails its schema and answers `line N`, N the 1-based number +// of the line as posted; a batch is one row per line with no blank lines, so +// the named line is batch[N-1]. +func refusedLine(reply []byte, batch []Row) int { + var refused struct { + Error string `json:"error"` + } + if json.Unmarshal(reply, &refused) != nil || refused.Error == "" { + return -1 + } + num, ok := strings.CutPrefix(refused.Error, "line ") + if !ok { + return -1 + } + colon := strings.IndexByte(num, ':') + if colon < 1 { + return -1 + } + line, err := strconv.Atoi(num[:colon]) + if err != nil || line < 1 || line > len(batch) { + return -1 + } + return line - 1 } // sendFile appends the rows to the file at path, creating it with mode 0600 From b6650ac58579c024e3b8a0864036eec41ef1801c Mon Sep 17 00:00:00 2001 From: santoshkumarradha Date: Thu, 17 Sep 2026 22:07:05 -0400 Subject: [PATCH 3/4] changes: a refused row leaves the outbox Co-Authored-By: codeaf --- .../unreleased/0000-a-refused-row-leaves-the-outbox.md | 9 +++++++++ 1 file changed, 9 insertions(+) create mode 100644 docs/changes/unreleased/0000-a-refused-row-leaves-the-outbox.md diff --git a/docs/changes/unreleased/0000-a-refused-row-leaves-the-outbox.md b/docs/changes/unreleased/0000-a-refused-row-leaves-the-outbox.md new file mode 100644 index 0000000000..9d61bc25b4 --- /dev/null +++ b/docs/changes/unreleased/0000-a-refused-row-leaves-the-outbox.md @@ -0,0 +1,9 @@ +--- +kind: fixed +title: A row the relay refuses by line leaves the outbox and the rows around it are sent +pr: 0000 +surface: [engine] +invalidates: + - "An outbox Send that met a refused batch left every row of it, and every row behind it, pending: the relay refuses a whole batch on the first line it cannot validate and answers `400 {\"error\":\"line N: \"}`, so one row the pool's schema would never accept — a payload written by an older or newer client, or schema drift — was retried at the head of the batch on every judged run and every start-up, refused again, and nothing from that install reached the pool again." + - "A 400 whose reply names `line N` now marks the named row dropped — the one marker that keeps it gone across a reopen — and posts the rest of the batch again within the same Send, one POST per refused line at the most and still inside the Send's budget. A 400 that names no row, a 413 and a 429 still answer an error and keep their rows pending: none of them names a row, so retiring one would be a guess, and the pending cap still drops rows from the old end when the retries would pile up. 5xx is unchanged." +--- \ No newline at end of file From 62b052438b80201613734f7018059a4f7ef05f6b Mon Sep 17 00:00:00 2001 From: santoshkumarradha Date: Thu, 17 Sep 2026 22:11:55 -0400 Subject: [PATCH 4/4] changes: stamp the refused-row note with its pull request Co-Authored-By: Claude Fable 5.1 --- ...-the-outbox.md => 1130-a-refused-row-leaves-the-outbox.md} | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) rename docs/changes/unreleased/{0000-a-refused-row-leaves-the-outbox.md => 1130-a-refused-row-leaves-the-outbox.md} (98%) diff --git a/docs/changes/unreleased/0000-a-refused-row-leaves-the-outbox.md b/docs/changes/unreleased/1130-a-refused-row-leaves-the-outbox.md similarity index 98% rename from docs/changes/unreleased/0000-a-refused-row-leaves-the-outbox.md rename to docs/changes/unreleased/1130-a-refused-row-leaves-the-outbox.md index 9d61bc25b4..3376b5b50d 100644 --- a/docs/changes/unreleased/0000-a-refused-row-leaves-the-outbox.md +++ b/docs/changes/unreleased/1130-a-refused-row-leaves-the-outbox.md @@ -1,9 +1,9 @@ --- kind: fixed title: A row the relay refuses by line leaves the outbox and the rows around it are sent -pr: 0000 +pr: 1130 surface: [engine] invalidates: - "An outbox Send that met a refused batch left every row of it, and every row behind it, pending: the relay refuses a whole batch on the first line it cannot validate and answers `400 {\"error\":\"line N: \"}`, so one row the pool's schema would never accept — a payload written by an older or newer client, or schema drift — was retried at the head of the batch on every judged run and every start-up, refused again, and nothing from that install reached the pool again." - "A 400 whose reply names `line N` now marks the named row dropped — the one marker that keeps it gone across a reopen — and posts the rest of the batch again within the same Send, one POST per refused line at the most and still inside the Send's budget. A 400 that names no row, a 413 and a 429 still answer an error and keep their rows pending: none of them names a row, so retiring one would be a guess, and the pending cap still drops rows from the old end when the retries would pile up. 5xx is unchanged." ---- \ No newline at end of file +---