perf: flush completion-tracked items on idle to bound Ack latency

ingest_sync_test.go

@@ -298,6 +298,123 @@ func TestIngestBatch_ClosedWhileServerFailsSettlesClosed(t *testing.T) {
 	}
 }
 
+// Core: a tracked item flushes as soon as the worker is idle, so IngestSync
+// returns well before maxDelay even when the batch size is never reached.
+func TestIngestSync_FlushesOnIdleBeforeMaxDelay(t *testing.T) {
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		io.Copy(io.Discard, r.Body)
+		w.WriteHeader(http.StatusOK)
+	}))
+	defer server.Close()
+
+	c := quickwit.NewClient(server.URL + "/api/v1/test")
+	c.SetConcurrent(1)
+	c.SetBatchSize(1000)            // size trigger never fires
+	c.SetMaxDelay(30 * time.Second) // ticker must not be what flushes it
+	defer c.Close()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	start := time.Now()
+	err := c.IngestSync(ctx, map[string]any{"index": 0})
+	elapsed := time.Since(start)
+
+	if err != nil {
+		t.Fatalf("IngestSync returned %v, want nil", err)
+	}
+	if elapsed > 2*time.Second {
+		t.Errorf("IngestSync took %v, want well under maxDelay (idle flush)", elapsed)
+	}
+}
+
+// Core: fire-and-forget Ingest is unaffected by idle flush — a partial batch
+// still waits for the timer rather than flushing immediately.
+func TestIngest_FireAndForgetDoesNotIdleFlush(t *testing.T) {
+	const maxDelay = 400 * time.Millisecond
+
+	arrived := make(chan time.Duration, 1)
+	start := time.Now()
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		select {
+		case arrived <- time.Since(start):
+		default:
+		}
+		io.Copy(io.Discard, r.Body)
+		w.WriteHeader(http.StatusOK)
+	}))
+	defer server.Close()
+
+	c := quickwit.NewClient(server.URL + "/api/v1/test")
+	c.SetConcurrent(1)
+	c.SetBatchSize(1000) // never reached
+	c.SetMaxDelay(maxDelay)
+	defer c.Close()
+
+	start = time.Now()
+	c.Ingest(
+		map[string]any{"index": 0},
+		map[string]any{"index": 1},
+	)
+
+	select {
+	case d := <-arrived:
+		// Must have waited for the timer, not flushed on idle. Use a margin
+		// below maxDelay to stay robust against scheduling jitter.
+		if d < maxDelay/2 {
+			t.Errorf("fire-and-forget flush arrived after %v, want it to wait ~%v for the timer", d, maxDelay)
+		}
+	case <-time.After(2 * time.Second):
+		t.Fatal("no request arrived before timeout")
+	}
+}
+
+// Core: even with idle flush active, a tracked burst still coalesces and every
+// record is delivered exactly once, in order.
+func TestIngestSync_IdleFlushPreservesDeliveryUnderBurst(t *testing.T) {
+	const numItems = 300
+
+	var mu sync.Mutex
+	var received []int
+
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		body, _ := io.ReadAll(r.Body)
+		mu.Lock()
+		received = append(received, parseIndices(body)...)
+		mu.Unlock()
+		w.WriteHeader(http.StatusOK)
+	}))
+	defer server.Close()
+
+	c := quickwit.NewClient(server.URL + "/api/v1/test")
+	c.SetConcurrent(1) // single worker => global order is observable
+	c.SetBatchSize(50)
+	c.SetMaxDelay(30 * time.Second)
+	defer c.Close()
+
+	// One call enqueues the whole burst, so the drain coalesces it into batches.
+	data := make([]any, numItems)
+	for i := range data {
+		data[i] = map[string]any{"index": i}
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+	if err := c.IngestSync(ctx, data...); err != nil {
+		t.Fatalf("IngestSync returned %v, want nil", err)
+	}
+
+	mu.Lock()
+	defer mu.Unlock()
+	if len(received) != numItems {
+		t.Fatalf("received %d items, want %d", len(received), numItems)
+	}
+	for i, idx := range received {
+		if idx != i {
+			t.Errorf("position %d: got index %d, want %d", i, idx, i)
+		}
+	}
+}
+
 // Core: IngestSync with no data is a no-op that returns nil even before setup.
 func TestIngestSync_EmptyIsNoop(t *testing.T) {
 	c := quickwit.NewClient("http://example/api/v1/test")

quickwit.go

@@ -353,9 +353,11 @@ func (c *Client) Ingest(data ...any) {
 // Delivery is at-least-once: there is an unavoidable window between the server's
 // 200 and the caller's Ack, so a crash can redeliver. Attach a deterministic
 // 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. For
-// low-volume topics, lower SetMaxDelay so a lone item is not held for up to the
-// flush interval before its first POST.
+// the subscription's ack-deadline; IngestSync never creates its own timeout.
+//
+// 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.
 //
 // With multiple values the result is a single first-error-wins verdict for the
 // whole call; ingest one document per call for a clean message-to-verdict map.
@@ -736,26 +738,79 @@ func (c *Client) loop() {
 		return false
 	}
 
+	// finalFlush drains the buffer on shutdown, settling/discarding anything the
+	// server never accepted.
+	finalFlush := func() {
+		if !retryFlush(true) {
+			slog.Error("quickwit: flush failed while closing")
+			for _, it := range buffer {
+				c.discardItem(it, &IngestError{Reason: ReasonClosed})
+			}
+		}
+	}
+
 	ticker := time.NewTicker(c.getMaxDelay())
 	defer ticker.Stop()
 
+	// hasTracked is true while the buffer may hold a completion-tracked item
+	// (one submitted via IngestSync / IngestBatch). Such items are latency
+	// sensitive — the caller is blocked waiting to Ack — so once one is buffered
+	// the worker flushes as soon as the channel goes idle instead of waiting for
+	// the ticker, bounding Ack latency by the round-trip rather than by maxDelay.
+	// Pure fire-and-forget traffic never sets this, so its batching is unchanged.
+	hasTracked := false
+
 	for {
 		select {
 		case <-ticker.C:
 			flushOversize()
+			if len(buffer) == 0 {
+				hasTracked = false
+			}
 		case x, ok := <-c.ingestBuffer:
 			if !ok { // channel closed
-				if !retryFlush(true) {
-					slog.Error("quickwit: flush failed while closing")
-					for _, it := range buffer {
-						c.discardItem(it, &IngestError{Reason: ReasonClosed})
+				finalFlush()
+				return
+			}
+			buffer = append(buffer, x)
+			if x.ack != nil {
+				hasTracked = true
+			}
+
+			if !hasTracked {
+				if len(buffer) >= batchSize {
+					retryFlush(false)
+				}
+				continue
+			}
+
+			// A tracked item is waiting: greedily pull whatever else is already
+			// queued (up to a full batch) so a burst still coalesces, then flush
+			// without waiting for the ticker.
+			closed := false
+		drain:
+			for len(buffer) < batchSize {
+				select {
+				case x, ok := <-c.ingestBuffer:
+					if !ok {
+						closed = true
+						break drain
+					}
+					buffer = append(buffer, x)
+					if x.ack != nil {
+						hasTracked = true
 					}
+				default:
+					break drain
 				}
+			}
+			if closed {
+				finalFlush()
 				return
 			}
-			buffer = append(buffer, x)
-			if len(buffer) >= batchSize {
-				retryFlush(false)
+			retryFlush(false)
+			if len(buffer) == 0 {
+				hasTracked = false
 			}
 		}
 	}