From b8568e6c76479488fa43b3d6e5db9e620bf510af Mon Sep 17 00:00:00 2001 From: Thanatat Tamtan Date: Sat, 13 Jun 2026 14:22:58 +0700 Subject: [PATCH] feat: detect server-side document rejections on a 200 ingest MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A Quickwit ingest 200 only means the documents were queued — its response body reports num_ingested_docs / num_rejected_docs, and a parse-rejected document (bad JSON or schema) is dropped while the client still saw 200. Today the client discards the body and Acks those documents as durable: silent loss (quickwit-oss/quickwit#5356, #3190). flush() now reads the response (bounded to 1 MiB) and inspects it: - Truncated/unreadable 200 -> retryable (ambiguous, don't settle). - num_rejected_docs > 0 with num_ingested_docs == 0 (whole batch rejected) -> settle tracked items with a new terminal ReasonRejected and discard fire-and-forget items. The verdict is exact because nothing was ingested. Thanks to idle-flush, a low-volume IngestSync document is flushed as its own one-doc batch, so the common pub/sub case gets a precise per-document verdict. - Partial rejection (some ingested, some rejected) can't be attributed to specific docs in a coalesced batch, so the batch is still Acked and the count is surfaced via a new OnReject(numRejected) callback + a warning log. - Absent/empty/unparseable body or missing fields (older servers) -> accepted, preserving the accept-on-200 behavior. Adds OnReject, a ReasonRejected DiscardReason, and an opt-in SetIngestDetailedResponse(true) that requests ?detailed_response=true so per-document parse-failure reasons are logged. Fire-and-forget batching, ordering, the failure taxonomy, and the no-double-settle invariant are unchanged (the all-rejected path settles each item once and drops the batch). Co-Authored-By: Claude Opus 4.8 (1M context) --- ingest_reject_test.go | 231 ++++++++++++++++++++++++++++++++++++++++++ quickwit.go | 150 +++++++++++++++++++++++++-- 2 files changed, 372 insertions(+), 9 deletions(-) create mode 100644 ingest_reject_test.go diff --git a/ingest_reject_test.go b/ingest_reject_test.go new file mode 100644 index 0000000..dbb0001 --- /dev/null +++ b/ingest_reject_test.go @@ -0,0 +1,231 @@ +package quickwit_test + +import ( + "context" + "errors" + "io" + "net/http" + "net/http/httptest" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/moonrhythm/quickwit" +) + +// Core: a 200 whose body reports the whole batch was parse-rejected +// (num_ingested_docs == 0) yields a terminal ReasonRejected, not a false nil. +func TestIngestSync_AllRejectedReturnsRejected(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + io.Copy(io.Discard, r.Body) + w.Header().Set("Content-Type", "application/json") + io.WriteString(w, `{"num_docs_for_processing":1,"num_ingested_docs":0,"num_rejected_docs":1}`) + })) + defer server.Close() + + var rejectedCount atomic.Int64 + c := quickwit.NewClient(server.URL + "/api/v1/test") + c.SetConcurrent(1) + c.OnReject(func(n int) { rejectedCount.Add(int64(n)) }) + defer c.Close() + + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + + err := c.IngestSync(ctx, map[string]any{"index": 0}) + var ie *quickwit.IngestError + if !errors.As(err, &ie) || ie.Reason != quickwit.ReasonRejected { + t.Fatalf("err = %v, want *IngestError{ReasonRejected}", err) + } + if n := rejectedCount.Load(); n != 1 { + t.Errorf("OnReject total = %d, want 1", n) + } +} + +// Core: a 200 that reports no rejection settles as accepted (nil). +func TestIngestSync_NoRejectionAccepts(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + io.Copy(io.Discard, r.Body) + io.WriteString(w, `{"num_docs_for_processing":1,"num_ingested_docs":1,"num_rejected_docs":0}`) + })) + defer server.Close() + + c := quickwit.NewClient(server.URL + "/api/v1/test") + c.SetConcurrent(1) + c.OnReject(func(int) { t.Error("OnReject must not fire when nothing was rejected") }) + defer c.Close() + + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + if err := c.IngestSync(ctx, map[string]any{"index": 0}); err != nil { + t.Fatalf("IngestSync returned %v, want nil", err) + } +} + +// Core: a partial rejection (some ingested, some rejected) cannot be attributed +// per-document, so the batch is still accepted (nil) but OnReject fires with the +// count — the operator's signal for the otherwise-silent loss. +func TestIngestSync_PartialRejectionAcksAndReports(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + io.Copy(io.Discard, r.Body) + io.WriteString(w, `{"num_docs_for_processing":3,"num_ingested_docs":2,"num_rejected_docs":1}`) + })) + defer server.Close() + + var reported atomic.Int64 + c := quickwit.NewClient(server.URL + "/api/v1/test") + c.SetConcurrent(1) + c.OnReject(func(n int) { reported.Store(int64(n)) }) + defer c.Close() + + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + // Three docs in one call -> one coalesced batch. + err := c.IngestSync(ctx, + map[string]any{"index": 0}, + map[string]any{"index": 1}, + map[string]any{"index": 2}, + ) + if err != nil { + t.Fatalf("partial rejection: IngestSync returned %v, want nil (cannot attribute)", err) + } + if n := reported.Load(); n != 1 { + t.Errorf("OnReject reported %d, want 1", n) + } +} + +// Core: an older/different server whose body lacks the rejection fields is +// treated as accepted (lenient), preserving the accept-on-200 contract. +func TestIngestSync_UnknownResponseBodyAccepts(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + io.Copy(io.Discard, r.Body) + io.WriteString(w, `{"num_docs_for_processing":1}`) // no num_rejected_docs + })) + defer server.Close() + + c := quickwit.NewClient(server.URL + "/api/v1/test") + c.SetConcurrent(1) + defer c.Close() + + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + if err := c.IngestSync(ctx, map[string]any{"index": 0}); err != nil { + t.Fatalf("IngestSync returned %v, want nil for a response without rejection fields", err) + } +} + +// Core: a truncated 200 body (transport cut mid-response) is ambiguous and must +// be retried, not settled. +func TestIngestSync_TruncatedResponseRetried(t *testing.T) { + var attempts atomic.Int64 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + io.Copy(io.Discard, r.Body) + n := attempts.Add(1) + if n == 1 { + // Promise more than we send, then hijack-close to truncate the body. + w.Header().Set("Content-Length", "1024") + w.WriteHeader(http.StatusOK) + io.WriteString(w, `{"num_ingested`) + if f, ok := w.(http.Flusher); ok { + f.Flush() + } + hj, ok := w.(http.Hijacker) + if ok { + conn, _, _ := hj.Hijack() + conn.Close() // abrupt close -> client read error + } + return + } + io.WriteString(w, `{"num_docs_for_processing":1,"num_ingested_docs":1,"num_rejected_docs":0}`) + })) + defer server.Close() + + c := quickwit.NewClient(server.URL + "/api/v1/test") + c.SetConcurrent(1) + c.SetBatchSize(1) + defer c.Close() + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + if err := c.IngestSync(ctx, map[string]any{"index": 0}); err != nil { + t.Fatalf("IngestSync returned %v, want nil after retrying the truncated response", err) + } + if n := attempts.Load(); n < 2 { + t.Errorf("attempts = %d, want >= 2 (truncated 200 retried)", n) + } +} + +// Core: SetIngestDetailedResponse adds ?detailed_response=true to the ingest URL. +func TestIngest_DetailedResponseQueryParam(t *testing.T) { + var mu sync.Mutex + var sawDetailed bool + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + mu.Lock() + if r.URL.Query().Get("detailed_response") == "true" { + sawDetailed = true + } + mu.Unlock() + io.Copy(io.Discard, r.Body) + io.WriteString(w, `{"num_ingested_docs":1,"num_rejected_docs":0}`) + })) + defer server.Close() + + c := quickwit.NewClient(server.URL + "/api/v1/test") + c.SetConcurrent(1) + c.SetBatchSize(1) + c.SetIngestDetailedResponse(true) + + c.Ingest(map[string]any{"index": 0}) + c.Close() + + mu.Lock() + defer mu.Unlock() + if !sawDetailed { + t.Error("ingest request did not carry ?detailed_response=true") + } +} + +// Core: with the whole batch rejected, fire-and-forget items are surfaced via +// OnDiscard (and the batch is dropped, not retried forever). +func TestIngest_AllRejectedDiscardsFireAndForget(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + body, _ := io.ReadAll(r.Body) + n := int64(strings.Count(strings.TrimSpace(string(body)), "\n") + 1) + io.WriteString(w, `{"num_ingested_docs":0,"num_rejected_docs":`+itoa(n)+`}`) + })) + defer server.Close() + + var discarded atomic.Int64 + c := quickwit.NewClient(server.URL + "/api/v1/test") + c.SetConcurrent(1) + c.SetBatchSize(2) + c.OnDiscard(func(any) { discarded.Add(1) }) + + c.Ingest(map[string]any{"index": 0}, map[string]any{"index": 1}) + + done := make(chan struct{}) + go func() { c.Close(); close(done) }() + select { + case <-done: + case <-time.After(5 * time.Second): + t.Fatal("Close did not return — worker wedged on rejection") + } + if n := discarded.Load(); n != 2 { + t.Errorf("OnDiscard fired %d times, want 2", n) + } +} + +func itoa(n int64) string { + if n == 0 { + return "0" + } + var b []byte + for n > 0 { + b = append([]byte{byte('0' + n%10)}, b...) + n /= 10 + } + return string(b) +} diff --git a/quickwit.go b/quickwit.go index cfca9f1..3772182 100644 --- a/quickwit.go +++ b/quickwit.go @@ -26,10 +26,37 @@ const ( ReduceBatchSizeToRatio = 0.9 // reduce 10% of the batch size ReduceBatchSizeMin = 0.1 // do not reduce below 10% of the default batch size ResetBatchSizeAfter = 10 * time.Minute + + // ingestResponseMaxBytes caps how much of the ingest response body is read + // before parsing, so a pathological body cannot exhaust memory. The default + // response is a few counts; only detailed_response with many parse failures + // approaches this. + ingestResponseMaxBytes = 1 << 20 // 1 MiB ) type OnDiscardFunc func(any) +// OnRejectFunc is called with the number of documents the server reported as +// parse-rejected in a single ingest response (a 200 with num_rejected_docs > 0). +// It is the only signal for a PARTIAL rejection, where the client cannot tell +// which documents in a coalesced batch failed and Acks the batch anyway. When a +// whole batch is rejected, tracked items also settle with ReasonRejected. +type OnRejectFunc func(numRejected int) + +// ingestResponse is the subset of the Quickwit ingest response the client acts +// on. Pointer fields distinguish "absent" (older servers) from zero. See +// https://quickwit.io/docs — a 200 only means the docs were queued. +type ingestResponse struct { + NumIngestedDocs *int64 `json:"num_ingested_docs"` + NumRejectedDocs *int64 `json:"num_rejected_docs"` + ParseFailures []ingestParseFailure `json:"parse_failures"` // only with detailed_response +} + +type ingestParseFailure struct { + Message string `json:"message"` + Reason string `json:"reason"` +} + // DiscardReason explains why a tracked ingest item (one submitted via IngestSync // or IngestBatch) was dropped before the server durably accepted it. It is // carried by *IngestError so a caller — e.g. a pub/sub handler — can decide @@ -52,6 +79,12 @@ const ( // dead-lettered and Acked; a misconfiguration must be fixed (Nacking would // redeliver forever). 5xx and 413 are not this — they stay retryable. ReasonServer + // ReasonRejected: the server returned 200 but its response reported that the + // document was parse-rejected (bad JSON or schema) and not indexed. Retrying + // never helps; dead-letter and Ack. Only reported when the whole flushed + // batch was rejected, so attribution is exact (see OnReject for the partial + // case). + ReasonRejected ) func (r DiscardReason) String() string { @@ -64,6 +97,8 @@ func (r DiscardReason) String() string { return "closed" case ReasonServer: return "server" + case ReasonRejected: + return "rejected" default: return "unknown" } @@ -201,8 +236,10 @@ type Client struct { stopWg sync.WaitGroup closeSignal chan struct{} onDiscard OnDiscardFunc + onReject OnRejectFunc autoReduceBatchSize bool gzipEnabled bool + detailedResponse bool sendMu sync.RWMutex // guards buffer sends against close(ingestBuffer) in Close closed bool // set under sendMu write lock in Close } @@ -261,6 +298,22 @@ func (c *Client) OnDiscard(f OnDiscardFunc) { c.onDiscard = f } +// OnReject registers a callback invoked with the number of documents the server +// reported as parse-rejected on an otherwise-successful (200) ingest. It is the +// hook for partial rejections that the client cannot attribute to a specific +// document. Set before the first Ingest. +func (c *Client) OnReject(f OnRejectFunc) { + c.onReject = f +} + +// SetIngestDetailedResponse requests Quickwit's detailed ingest response +// (?detailed_response=true) so per-document parse-failure reasons are logged +// when documents are rejected. It adds response size/CPU on the server, so it is +// off by default. Set before the first Ingest. +func (c *Client) SetIngestDetailedResponse(enabled bool) { + c.detailedResponse = enabled +} + func (c *Client) httpClient() *http.Client { if c.client != nil { return c.client @@ -377,6 +430,51 @@ func (c *Client) invokeOnDiscard(data any) { } } +func (c *Client) invokeOnReject(numRejected int) { + if c.onReject != nil { + c.onReject(numRejected) + } +} + +// inspectIngestResponse parses a 200 ingest response and reports whether the +// server rejected any documents and whether it rejected ALL of them. It is +// lenient: an unparseable body or absent fields (older servers) is treated as +// "no rejection info" so the accept-on-200 behavior is preserved. allRejected is +// reported only when the server ingested nothing, which makes the per-item +// ReasonRejected verdict exact (no accepted doc is dead-lettered). +func (c *Client) inspectIngestResponse(body []byte, endpoint string) (rejected, allRejected bool) { + // An empty body (some proxies, older servers) carries no rejection info; + // accept silently rather than warn on every flush. + if len(bytes.TrimSpace(body)) == 0 { + return false, false + } + var r ingestResponse + if err := json.Unmarshal(body, &r); err != nil { + slog.Warn("quickwit: could not parse ingest response", "endpoint", endpoint, "error", err) + return false, false + } + if r.NumRejectedDocs == nil || *r.NumRejectedDocs <= 0 { + return false, false + } + + ingested := int64(-1) // -1: server did not report it + if r.NumIngestedDocs != nil { + ingested = *r.NumIngestedDocs + } + slog.Warn("quickwit: server rejected documents", + "endpoint", endpoint, + "rejected", *r.NumRejectedDocs, + "ingested", ingested, + ) + for _, pf := range r.ParseFailures { + slog.Warn("quickwit: document parse failure", "reason", pf.Reason, "message", pf.Message) + } + c.invokeOnReject(int(*r.NumRejectedDocs)) + + // Unambiguous only when the server ingested nothing: every doc was rejected. + return true, r.NumIngestedDocs != nil && *r.NumIngestedDocs == 0 +} + // Ingest enqueues data for asynchronous, fire-and-forget delivery. Each value is // JSON-encoded by a background worker and sent in batches. Ingest does not report // delivery: on a crash before the batch is flushed the data is lost, and after @@ -416,10 +514,13 @@ func (c *Client) Ingest(data ...any) { // value returns *IngestError{Reason: ReasonEncode} immediately, before anything // is buffered — enqueues the items, and blocks until: // -// - all items received HTTP 200 from {endpoint}/ingest → returns nil (Ack); -// - an item was terminally dropped → returns an *IngestError with ReasonEncode -// (poison: dead-letter and Ack), or ReasonBufferFull / ReasonClosed -// (transient: Nack to redeliver); +// - all items were durably accepted (HTTP 200 with no rejection) → returns nil +// (Ack); +// - an item was terminally dropped → returns an *IngestError whose Reason is +// ReasonEncode (bad document) or ReasonRejected (server parse-rejected the +// whole batch) — poison, dead-letter and Ack; ReasonServer (permanent 4xx — +// inspect, usually a misconfig to fix); or ReasonBufferFull / ReasonClosed +// (transient — Nack to redeliver); // - ctx fired first → returns a wrapped context error. This is AMBIGUOUS: the // item is still buffered and may yet be ingested, so treat it as "not // confirmed" and Nack. @@ -605,6 +706,9 @@ func (c *Client) loop() { endpoint := c.endpoint endpoint = strings.TrimSuffix(endpoint, "/") endpoint = endpoint + "/ingest" + if c.detailedResponse { + endpoint = endpoint + "?detailed_response=true" + } flush := func(batch []ingestItem) bool { if len(batch) == 0 { @@ -677,10 +781,11 @@ func (c *Client) loop() { if err != nil { return false } - io.Copy(io.Discard, resp.Body) - resp.Body.Close() if resp.StatusCode != http.StatusOK { + io.Copy(io.Discard, resp.Body) + resp.Body.Close() + slog.Error("quickwit: ingest status not ok", "status", resp.Status) if resp.StatusCode == http.StatusRequestEntityTooLarge { @@ -731,9 +836,36 @@ func (c *Client) loop() { return false } - // HTTP 200 is the only durable-acceptance point: settle every item in the - // batch (settle is a no-op for fire-and-forget and encode-failure items, - // which have ack == nil), then discard items that failed encoding. + // HTTP 200 only means the documents were queued. Read the response (bounded) + // and inspect it: the server reports how many documents it parse-rejected. + respBody, readErr := io.ReadAll(io.LimitReader(resp.Body, ingestResponseMaxBytes)) + io.Copy(io.Discard, resp.Body) // drain any remainder so the connection can be reused + resp.Body.Close() + if readErr != nil { + // A truncated/incomplete 200 is ambiguous — retry rather than settle. + slog.Error("quickwit: failed to read ingest response", "error", readErr) + return false + } + + // When the whole batch was parse-rejected the verdict is exact, so settle + // tracked items with ReasonRejected (and discard fire-and-forget) instead + // of falsely reporting them durable. A partial rejection cannot be + // attributed to specific documents in a coalesced batch — it is surfaced + // via OnReject inside inspectIngestResponse and the batch is still Acked. + if rejected, allRejected := c.inspectIngestResponse(respBody, endpoint); rejected && allRejected { + err := &IngestError{ + Reason: ReasonRejected, + Err: fmt.Errorf("quickwit: server rejected all %d document(s) in the batch", len(batch)), + } + for _, it := range batch { + c.discardItem(it, err) + } + return true + } + + // Durably accepted: settle every item in the batch (settle is a no-op for + // fire-and-forget and encode-failure items, which have ack == nil), then + // discard items that failed encoding. for _, it := range batch { it.ack.settle(nil) }