fix: stop row isolation after ambiguous send
This commit is contained in:
+9
-1
@@ -431,12 +431,20 @@ func Flush(ctx context.Context, conn BatchPreparer, entries []*LogEntry) FlushRe
|
|||||||
}
|
}
|
||||||
|
|
||||||
combined := FlushResult{}
|
combined := FlushResult{}
|
||||||
for _, entry := range entries {
|
for i, entry := range entries {
|
||||||
single := flushOnce(ctx, conn, []*LogEntry{entry}).public([]*LogEntry{entry})
|
single := flushOnce(ctx, conn, []*LogEntry{entry}).public([]*LogEntry{entry})
|
||||||
combined.Retry = append(combined.Retry, single.Retry...)
|
combined.Retry = append(combined.Retry, single.Retry...)
|
||||||
combined.Failed += single.Failed
|
combined.Failed += single.Failed
|
||||||
combined.Ambiguous += single.Ambiguous
|
combined.Ambiguous += single.Ambiguous
|
||||||
combined.Err = errors.Join(combined.Err, single.Err)
|
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
|
return combined
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -42,8 +42,11 @@ type fakePool struct {
|
|||||||
queries []string
|
queries []string
|
||||||
batches []*fakeBatch
|
batches []*fakeBatch
|
||||||
scripts []batchScript
|
scripts []batchScript
|
||||||
|
unhealthy bool
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func (p *fakePool) MarkUnhealthy() { p.unhealthy = true }
|
||||||
|
|
||||||
type queueBackend struct {
|
type queueBackend struct {
|
||||||
prepare func(context.Context, string) (logger.Batch, error)
|
prepare func(context.Context, string) (logger.Batch, error)
|
||||||
healthy atomic.Bool
|
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) {
|
func TestEstimatedBytesCoversStringsAndByteSlices(t *testing.T) {
|
||||||
entry := &logger.LogEntry{
|
entry := &logger.LogEntry{
|
||||||
RequestID: "1", Method: "22", Path: "333", Query: "4444", ClientIP: "55555", Error: "666666",
|
RequestID: "1", Method: "22", Path: "333", Query: "4444", ClientIP: "55555", Error: "666666",
|
||||||
|
|||||||
Reference in New Issue
Block a user