diff --git a/ingest_sync_test.go b/ingest_sync_test.go index ce5ec22..ebe0e5c 100644 --- a/ingest_sync_test.go +++ b/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") diff --git a/quickwit.go b/quickwit.go index e5a3aa2..5cde4f3 100644 --- a/quickwit.go +++ b/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 } } }