feat: opt-in commit=wait_for for completion-tracked batches

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
+}

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)),