diff --git a/logger/queue.go b/logger/queue.go index 43da90d..7d22e6d 100644 --- a/logger/queue.go +++ b/logger/queue.go @@ -431,12 +431,20 @@ func Flush(ctx context.Context, conn BatchPreparer, entries []*LogEntry) FlushRe } combined := FlushResult{} - for _, entry := range entries { + for i, entry := range entries { single := flushOnce(ctx, conn, []*LogEntry{entry}).public([]*LogEntry{entry}) combined.Retry = append(combined.Retry, single.Retry...) combined.Failed += single.Failed combined.Ambiguous += single.Ambiguous combined.Err = errors.Join(combined.Err, single.Err) + if single.Ambiguous > 0 { + if backend, ok := conn.(interface{ MarkUnhealthy() }); ok { + backend.MarkUnhealthy() + } + combined.Failed += len(combined.Retry) + len(entries) - i - 1 + combined.Retry = nil + break + } } return combined } diff --git a/tests/logger/queue_test.go b/tests/logger/queue_test.go index 40374f7..2195340 100644 --- a/tests/logger/queue_test.go +++ b/tests/logger/queue_test.go @@ -38,12 +38,15 @@ func (f *fakeBatch) Send() error { return f.script.sendErr } func (f *fakeBatch) Abort() error { f.abortCalls++; return nil } type fakePool struct { - calls int - queries []string - batches []*fakeBatch - scripts []batchScript + calls int + queries []string + batches []*fakeBatch + scripts []batchScript + unhealthy bool } +func (p *fakePool) MarkUnhealthy() { p.unhealthy = true } + type queueBackend struct { prepare func(context.Context, string) (logger.Batch, error) healthy atomic.Bool @@ -215,6 +218,26 @@ func TestFlushAppendRecoveryCountsAmbiguousSingleSend(t *testing.T) { } } +func TestFlushAppendRecoveryStopsAfterFirstAmbiguousSend(t *testing.T) { + p := &fakePool{scripts: []batchScript{ + {appendAt: 1, appendErr: errors.New("bad batch")}, + {sendErr: errors.New("ack lost")}, + {}, + }} + entries := []*logger.LogEntry{{RequestID: "a"}, {RequestID: "b"}, {RequestID: "c"}} + + result := logger.Flush(context.Background(), p, entries) + if result.Failed != 3 || result.Ambiguous != 1 || len(result.Retry) != 0 { + t.Fatalf("unexpected result: %+v", result) + } + if p.calls != 2 { + t.Fatalf("PrepareBatch calls=%d want 2; entries after ambiguous Send must not be attempted", p.calls) + } + if !p.unhealthy { + t.Fatal("ambiguous singleton Send did not immediately mark backend unhealthy") + } +} + func TestEstimatedBytesCoversStringsAndByteSlices(t *testing.T) { entry := &logger.LogEntry{ RequestID: "1", Method: "22", Path: "333", Query: "4444", ClientIP: "55555", Error: "666666",