From c868dd474358ad1afc42c0245cbf9ce4a2b5db56 Mon Sep 17 00:00:00 2001 From: Thanatat Tamtan Date: Sat, 13 Jun 2026 14:47:33 +0700 Subject: [PATCH] feat: opt-in commit=wait_for for completion-tracked batches A 200 from the ingest endpoint means the documents reached the server's WAL, not that they are committed/searchable. For tracked callers (IngestSync / IngestBatch) settle(nil) is a durability claim, so make the commit boundary configurable. Add SetIngestCommit(CommitAuto|CommitWaitFor|CommitForce). The mode is applied per batch: only when a batch carries a tracked item does it use the stronger commit (?commit=wait_for / force); pure fire-and-forget batches keep commit=auto and pay no extra latency. Two endpoint URLs are precomputed per worker, so the default (CommitAuto) path is unchanged. Combines cleanly with ?detailed_response=true. wait_for/force block the request until commit, which can exceed the default ResponseHeaderTimeout (ingestTimeout/3); the IngestSync and SetIngestCommit docs spell out the >3x-commit-interval timeout headroom needed to avoid timeout-driven retries/duplicates. Also tightened the IngestSync doc: nil means WAL-persisted (crash-safe) by default, committed/searchable under wait_for. Co-Authored-By: Claude Opus 4.8 (1M context) --- commit_mode_test.go | 125 ++++++++++++++++++++++++++++++++++++++++++++ quickwit.go | 96 +++++++++++++++++++++++++++++++--- 2 files changed, 213 insertions(+), 8 deletions(-) create mode 100644 commit_mode_test.go diff --git a/commit_mode_test.go b/commit_mode_test.go new file mode 100644 index 0000000..8ab0a0c --- /dev/null +++ b/commit_mode_test.go @@ -0,0 +1,125 @@ +package quickwit_test + +import ( + "context" + "io" + "net/http" + "net/http/httptest" + "net/url" + "sync" + "testing" + "time" + + "github.com/moonrhythm/quickwit" +) + +// queryCapture records the query string of the first ingest request. +func queryCapture() (http.HandlerFunc, func() string) { + var mu sync.Mutex + var q string + var got bool + h := func(w http.ResponseWriter, r *http.Request) { + mu.Lock() + if !got { + q = r.URL.RawQuery + got = true + } + mu.Unlock() + io.Copy(io.Discard, r.Body) + io.WriteString(w, `{"num_ingested_docs":1,"num_rejected_docs":0}`) + } + return h, func() string { mu.Lock(); defer mu.Unlock(); return q } +} + +// Core: by default no commit parameter is sent (commit=auto behavior). +func TestIngest_CommitAutoByDefault(t *testing.T) { + h, query := queryCapture() + server := httptest.NewServer(h) + defer server.Close() + + c := quickwit.NewClient(server.URL + "/api/v1/test") + c.SetConcurrent(1) + c.SetBatchSize(1) + c.Ingest(map[string]any{"index": 0}) + c.Close() + + if q := query(); q != "" { + t.Errorf("query = %q, want empty (no commit param by default)", q) + } +} + +// Core: a tracked batch carries the configured commit mode. +func TestIngestSync_CommitWaitForOnTracked(t *testing.T) { + h, query := queryCapture() + server := httptest.NewServer(h) + defer server.Close() + + c := quickwit.NewClient(server.URL + "/api/v1/test") + c.SetConcurrent(1) + c.SetIngestCommit(quickwit.CommitWaitFor) + 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) + } + + if q := query(); !hasParam(q, "commit", "wait_for") { + t.Errorf("query = %q, want commit=wait_for", q) + } +} + +// Core: a fire-and-forget batch keeps commit=auto even when a tracked commit +// mode is configured — it carries no tracked item, so it must not pay the cost. +func TestIngest_FireAndForgetUsesAutoWithWaitForSet(t *testing.T) { + h, query := queryCapture() + server := httptest.NewServer(h) + defer server.Close() + + c := quickwit.NewClient(server.URL + "/api/v1/test") + c.SetConcurrent(1) + c.SetBatchSize(1) + c.SetIngestCommit(quickwit.CommitWaitFor) + c.Ingest(map[string]any{"index": 0}) // fire-and-forget => no tracked item + c.Close() + + if q := query(); hasParam(q, "commit", "wait_for") { + t.Errorf("query = %q, want no commit param for a fire-and-forget batch", q) + } +} + +// Core: commit and detailed_response combine into one well-formed query. +func TestIngestSync_CommitAndDetailedResponseCombine(t *testing.T) { + h, query := queryCapture() + server := httptest.NewServer(h) + defer server.Close() + + c := quickwit.NewClient(server.URL + "/api/v1/test") + c.SetConcurrent(1) + c.SetIngestCommit(quickwit.CommitForce) + c.SetIngestDetailedResponse(true) + 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) + } + + q := query() + if !hasParam(q, "commit", "force") { + t.Errorf("query = %q, want commit=force", q) + } + if !hasParam(q, "detailed_response", "true") { + t.Errorf("query = %q, want detailed_response=true", q) + } +} + +func hasParam(rawQuery, key, val string) bool { + v, err := url.ParseQuery(rawQuery) + if err != nil { + return false + } + return v.Get(key) == val +} diff --git a/quickwit.go b/quickwit.go index 3772182..737ddd7 100644 --- a/quickwit.go +++ b/quickwit.go @@ -36,6 +36,34 @@ const ( type OnDiscardFunc func(any) +// CommitMode selects Quickwit's ingest commit behavior (the ?commit= query +// parameter). It controls what a 200 response means for durability/visibility. +type CommitMode int + +const ( + // CommitAuto (default): the server returns once documents are persisted to + // its ingest write-ahead log. Crash-safe, but not necessarily searchable yet. + CommitAuto CommitMode = iota + // CommitWaitFor: the server returns only once the documents are committed and + // searchable (read-your-writes). Higher latency — the request blocks until the + // next commit — so the ingest timeout must be large enough to cover it. + CommitWaitFor + // CommitForce: like wait_for but also forces an immediate commit. Rarely + // wanted; a commit per flush creates many tiny splits and hurts indexing. + CommitForce +) + +func (m CommitMode) queryValue() string { + switch m { + case CommitWaitFor: + return "wait_for" + case CommitForce: + return "force" + default: + return "" + } +} + // 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 @@ -104,6 +132,22 @@ func (r DiscardReason) String() string { } } +// ingestURL builds the ingest endpoint with optional query parameters. base must +// have no query string (the {endpoint}/ingest convention). +func ingestURL(base, commit string, detailed bool) string { + var params []string + if commit != "" { + params = append(params, "commit="+commit) + } + if detailed { + params = append(params, "detailed_response=true") + } + if len(params) == 0 { + return base + } + return base + "?" + strings.Join(params, "&") +} + // permanentIngestStatus reports whether an HTTP status from the ingest endpoint // is a permanent rejection that retrying the same batch cannot fix. 5xx, 408, // 425, 429 and 413 are deliberately excluded — they stay retryable (413 has its @@ -240,6 +284,7 @@ type Client struct { autoReduceBatchSize bool gzipEnabled bool detailedResponse bool + commitMode CommitMode sendMu sync.RWMutex // guards buffer sends against close(ingestBuffer) in Close closed bool // set under sendMu write lock in Close } @@ -314,6 +359,24 @@ func (c *Client) SetIngestDetailedResponse(enabled bool) { c.detailedResponse = enabled } +// SetIngestCommit selects the commit mode for batches that contain a +// completion-tracked item (submitted via IngestSync / IngestBatch), so a +// tracked Ack can mean "committed" rather than just "queued". The default, +// CommitAuto, is unchanged behavior. Fire-and-forget-only batches always use +// auto (no latency cost); a mixed batch uses the stronger mode if any tracked +// item is present. +// +// CommitWaitFor / CommitForce make the ingest request block until commit, which +// can take until the index's next commit. The timeouts must cover that: the +// default transport's ResponseHeaderTimeout is getIngestTimeout()/3, so set +// SetIngestTimeout to comfortably more than 3× the index commit interval (or +// supply your own http.Client via SetHTTPClient). Otherwise the request times +// out, retries, and may double-ingest — rely on a dedup id. Set before the first +// Ingest. +func (c *Client) SetIngestCommit(mode CommitMode) { + c.commitMode = mode +} + func (c *Client) httpClient() *http.Client { if c.client != nil { return c.client @@ -530,6 +593,11 @@ func (c *Client) Ingest(data ...any) { // document id and dedup on it. Pass a ctx whose deadline sits comfortably inside // the subscription's ack-deadline; IngestSync never creates its own timeout. // +// By default a nil result means the documents are persisted to the server's +// ingest write-ahead log (crash-safe) but not necessarily searchable yet. Call +// SetIngestCommit(CommitWaitFor) if Ack should mean committed and searchable, at +// the cost of higher per-flush latency. +// // Ack latency is bounded by the round-trip, not by maxDelay: because the items // are completion-tracked, the worker flushes them as soon as it goes idle rather // than waiting out the flush interval, while still coalescing bursts into batches. @@ -703,12 +771,13 @@ func (c *Client) loop() { // reset at the top of every flush so it only reflects the most recent attempt. var retryAfter time.Duration - endpoint := c.endpoint - endpoint = strings.TrimSuffix(endpoint, "/") - endpoint = endpoint + "/ingest" - if c.detailedResponse { - endpoint = endpoint + "?detailed_response=true" - } + base := strings.TrimSuffix(c.endpoint, "/") + "/ingest" + // Two precomputed URLs: the default (commit=auto, used for fire-and-forget) + // and the tracked variant carrying the configured commit mode. They are equal + // when commitMode is CommitAuto, so the common path is unchanged. + endpointDefault := ingestURL(base, "", c.detailedResponse) + endpointTracked := ingestURL(base, c.commitMode.queryValue(), c.detailedResponse) + useTrackedCommit := endpointTracked != endpointDefault flush := func(batch []ingestItem) bool { if len(batch) == 0 { @@ -724,7 +793,11 @@ func (c *Client) loop() { // for the ack==nil encode failures — so no separate "encoded" slice is // allocated per flush. var encodeFailures []ingestItem + batchHasTracked := false for _, it := range batch { + if it.ack != nil { + batchHasTracked = true + } if it.raw != nil { buf.Write(it.raw) continue @@ -736,6 +809,13 @@ func (c *Client) loop() { } } + // Use the configured commit mode only when this batch carries a tracked + // item; fire-and-forget batches keep commit=auto (no latency cost). + reqURL := endpointDefault + if useTrackedCommit && batchHasTracked { + reqURL = endpointTracked + } + // All items were unencodable — discard them and report success so the // caller clears the buffer and does not retry with the same items. if buf.Len() == 0 { @@ -768,7 +848,7 @@ func (c *Client) loop() { ctx, cancel = context.WithTimeout(ctx, t) } defer cancel() - req, err := http.NewRequestWithContext(ctx, http.MethodPost, endpoint, bytes.NewReader(body)) + req, err := http.NewRequestWithContext(ctx, http.MethodPost, reqURL, bytes.NewReader(body)) if err != nil { return false } @@ -852,7 +932,7 @@ func (c *Client) loop() { // 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 { + if rejected, allRejected := c.inspectIngestResponse(respBody, reqURL); rejected && allRejected { err := &IngestError{ Reason: ReasonRejected, Err: fmt.Errorf("quickwit: server rejected all %d document(s) in the batch", len(batch)),