npc0-hue
e73fe765c3
A segment that the platform already acknowledged could still be extended by the next tailed line, so its checksum covered entries stored under a different batch. The platform then rejected the same body every second while RejectStreamAfter skipped it, because it only quarantined segments that start after the platform latest. Every newly allocated line was quarantined by the following recovery pass, so live log ingest never resumed. Quarantine pending segments that reach beyond the acknowledged range, never extend a segment the platform already stored, and log the gap recovery so a stalled stream is diagnosable from the spool alone.
487 lines
19 KiB
Go
487 lines
19 KiB
Go
package spool
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"os"
|
|
"path/filepath"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
|
|
"browser.local/run/protocol"
|
|
)
|
|
|
|
func TestLogSpoolAggregatesContiguousEntriesForOneStream(t *testing.T) {
|
|
logSpool, err := NewLogSpool(t.TempDir())
|
|
if err != nil {
|
|
t.Fatalf("new log spool: %v", err)
|
|
}
|
|
checksum := func(entries []protocol.LogEntry) (string, error) { return fmt.Sprintf("sha256:%d", len(entries)), nil }
|
|
for sequence := uint64(1); sequence <= 3; sequence++ {
|
|
batch := validSpoolLogBatch(sequence, sequence)
|
|
if err := logSpool.EnqueueAggregated(batch, checksum); err != nil {
|
|
t.Fatalf("enqueue sequence %d: %v", sequence, err)
|
|
}
|
|
}
|
|
pending, err := logSpool.Pending()
|
|
if err != nil {
|
|
t.Fatalf("pending: %v", err)
|
|
}
|
|
if len(pending) != 1 || pending[0].FirstSeq != 1 || pending[0].LastSeq != 3 || len(pending[0].Entries) != 3 || pending[0].Checksum != "sha256:3" {
|
|
t.Fatalf("expected one aggregated batch, got %+v", pending)
|
|
}
|
|
if err := logSpool.Ack(protocol.LogBatchIngestResponse{LogStreamID: "log-1", AcceptedFrom: 1, AcceptedTo: 3}); err != nil {
|
|
t.Fatalf("ack aggregated batch: %v", err)
|
|
}
|
|
pending, err = logSpool.Pending()
|
|
if err != nil || len(pending) != 0 {
|
|
t.Fatalf("expected acknowledged aggregation removed, pending=%+v err=%v", pending, err)
|
|
}
|
|
}
|
|
|
|
func TestLogSpoolRetainsPendingAndRemovesAcknowledgedBatch(t *testing.T) {
|
|
spool, err := NewLogSpool(t.TempDir())
|
|
if err != nil {
|
|
t.Fatalf("new log spool: %v", err)
|
|
}
|
|
first := validSpoolLogBatch(1, 2)
|
|
second := validSpoolLogBatch(3, 3)
|
|
if err := spool.Enqueue(first); err != nil {
|
|
t.Fatalf("enqueue first: %v", err)
|
|
}
|
|
if err := spool.Enqueue(second); err != nil {
|
|
t.Fatalf("enqueue second: %v", err)
|
|
}
|
|
pending, err := spool.Pending()
|
|
if err != nil {
|
|
t.Fatalf("pending before ack: %v", err)
|
|
}
|
|
if len(pending) != 2 {
|
|
t.Fatalf("expected two pending batches, got %+v", pending)
|
|
}
|
|
|
|
if err := spool.Ack(protocol.LogBatchIngestResponse{LogStreamID: "log-1", AcceptedFrom: 1, AcceptedTo: 2}); err != nil {
|
|
t.Fatalf("ack first: %v", err)
|
|
}
|
|
pending, err = spool.Pending()
|
|
if err != nil {
|
|
t.Fatalf("pending after ack: %v", err)
|
|
}
|
|
if len(pending) != 1 || pending[0].FirstSeq != 3 {
|
|
t.Fatalf("expected second batch pending, got %+v", pending)
|
|
}
|
|
}
|
|
|
|
func TestLogSpoolRetainsBatchWhenAckDoesNotCoverRange(t *testing.T) {
|
|
spool, err := NewLogSpool(t.TempDir())
|
|
if err != nil {
|
|
t.Fatalf("new log spool: %v", err)
|
|
}
|
|
if err := spool.Enqueue(validSpoolLogBatch(1, 2)); err != nil {
|
|
t.Fatalf("enqueue: %v", err)
|
|
}
|
|
if err := spool.Ack(protocol.LogBatchIngestResponse{LogStreamID: "log-1", AcceptedFrom: 1, AcceptedTo: 1}); err != nil {
|
|
t.Fatalf("partial ack: %v", err)
|
|
}
|
|
pending, err := spool.Pending()
|
|
if err != nil {
|
|
t.Fatalf("pending: %v", err)
|
|
}
|
|
if len(pending) != 1 {
|
|
t.Fatalf("expected batch to remain pending, got %+v", pending)
|
|
}
|
|
}
|
|
|
|
func TestLogSpoolQuarantinesPermanentRejectedBatch(t *testing.T) {
|
|
root := t.TempDir()
|
|
logSpool, err := NewLogSpool(root)
|
|
if err != nil {
|
|
t.Fatalf("new log spool: %v", err)
|
|
}
|
|
if err := logSpool.Enqueue(validSpoolLogBatch(5, 5)); err != nil {
|
|
t.Fatalf("enqueue: %v", err)
|
|
}
|
|
flushed, err := logSpool.Flush(context.Background(), permanentRejectLogBatchClient{})
|
|
if err != nil {
|
|
t.Fatalf("flush permanent rejection: %v", err)
|
|
}
|
|
if flushed != 1 {
|
|
t.Fatalf("expected one quarantined batch, got %d", flushed)
|
|
}
|
|
pending, err := logSpool.Pending()
|
|
if err != nil {
|
|
t.Fatalf("pending: %v", err)
|
|
}
|
|
if len(pending) != 0 {
|
|
t.Fatalf("expected no pending batches, got %+v", pending)
|
|
}
|
|
rejected, err := os.ReadDir(filepath.Join(root, "logs-rejected"))
|
|
if err != nil {
|
|
t.Fatalf("read rejected dir: %v", err)
|
|
}
|
|
if len(rejected) != 1 || !filepath.IsLocal(rejected[0].Name()) {
|
|
t.Fatalf("expected one local rejected file, got %+v", rejected)
|
|
}
|
|
}
|
|
|
|
func TestLogSpoolFlushPrioritizesNewerProcessSessions(t *testing.T) {
|
|
logSpool, err := NewLogSpool(t.TempDir())
|
|
if err != nil {
|
|
t.Fatalf("new log spool: %v", err)
|
|
}
|
|
oldBatch := validSpoolLogBatch(1, 1)
|
|
oldBatch.LogStreamID = "run.endpoint.server.a-old.stdout"
|
|
oldBatch.LogSessionID = "a-old"
|
|
oldBatch.SessionStartedAt = time.Date(2026, 7, 3, 12, 0, 0, 0, time.UTC)
|
|
newBatch := validSpoolLogBatch(1, 1)
|
|
newBatch.LogStreamID = "run.endpoint.server.z-new.stdout"
|
|
newBatch.LogSessionID = "z-new"
|
|
newBatch.SessionStartedAt = time.Date(2026, 7, 3, 13, 0, 0, 0, time.UTC)
|
|
newBatchNext := newBatch
|
|
newBatchNext.FirstSeq = 2
|
|
newBatchNext.LastSeq = 2
|
|
newBatchNext.Entries = []protocol.LogEntry{{Seq: 2, Timestamp: time.Date(2026, 7, 3, 13, 0, 2, 0, time.UTC), Level: "info", Line: "line 2"}}
|
|
for _, batch := range []protocol.LogBatchIngestRequest{oldBatch, newBatch, newBatchNext} {
|
|
if err := logSpool.Enqueue(batch); err != nil {
|
|
t.Fatalf("enqueue %s:%d: %v", batch.LogStreamID, batch.FirstSeq, err)
|
|
}
|
|
}
|
|
client := &recordingLogBatchClient{}
|
|
flushed, err := logSpool.Flush(context.Background(), client)
|
|
if err != nil || flushed != 3 {
|
|
t.Fatalf("flush: flushed=%d err=%v", flushed, err)
|
|
}
|
|
if len(client.batches) != 3 || client.batches[0].LogStreamID != newBatch.LogStreamID || client.batches[0].FirstSeq != 1 || client.batches[1].LogStreamID != newBatch.LogStreamID || client.batches[1].FirstSeq != 2 || client.batches[2].LogStreamID != oldBatch.LogStreamID {
|
|
t.Fatalf("expected new session first while preserving per-stream order, got %+v", client.batches)
|
|
}
|
|
}
|
|
|
|
func TestLogSpoolSequenceGapResetsWatermarkToPlatformLatest(t *testing.T) {
|
|
logSpool, err := NewLogSpool(t.TempDir())
|
|
if err != nil {
|
|
t.Fatalf("new log spool: %v", err)
|
|
}
|
|
if err := logSpool.Enqueue(validSpoolLogBatch(10, 10)); err != nil {
|
|
t.Fatalf("enqueue gapped batch: %v", err)
|
|
}
|
|
if err := logSpool.Enqueue(validSpoolLogBatch(11, 11)); err != nil {
|
|
t.Fatalf("enqueue later gapped batch: %v", err)
|
|
}
|
|
client := &recoveringSequenceGapLogBatchClient{latestSeq: 5}
|
|
flushed, err := logSpool.Flush(context.Background(), client)
|
|
if err != nil || flushed != 2 {
|
|
t.Fatalf("recover sequence gap: flushed=%d err=%v", flushed, err)
|
|
}
|
|
if len(client.progressBatches) != 1 || client.progressBatches[0].FirstSeq != 10 {
|
|
t.Fatalf("expected one progress recovery from gapped batch, got %+v", client.progressBatches)
|
|
}
|
|
pending, err := logSpool.Pending()
|
|
if err != nil || len(pending) != 0 {
|
|
t.Fatalf("expected gapped batches rejected, pending=%+v err=%v", pending, err)
|
|
}
|
|
checksum := func(entries []protocol.LogEntry) (string, error) { return fmt.Sprintf("sha256:%d", len(entries)), nil }
|
|
next := validSpoolLogBatch(0, 0)
|
|
sequence, appended, err := logSpool.EnqueueNextAggregated(context.Background(), next, nil, nil, checksum)
|
|
if err != nil || !appended || sequence != 6 {
|
|
t.Fatalf("expected next sequence to resume after platform latest, sequence=%d appended=%t err=%v", sequence, appended, err)
|
|
}
|
|
}
|
|
|
|
func TestLogSpoolSequenceGapRejectsStraddlingAcknowledgedSegment(t *testing.T) {
|
|
root := t.TempDir()
|
|
logSpool, err := NewLogSpool(root)
|
|
if err != nil {
|
|
t.Fatalf("new log spool: %v", err)
|
|
}
|
|
// The acknowledged range ends inside this segment: sequences 10..13 were
|
|
// extended after the platform had already stored sequence 10.
|
|
if err := logSpool.Enqueue(validSpoolLogBatch(10, 13)); err != nil {
|
|
t.Fatalf("enqueue straddling batch: %v", err)
|
|
}
|
|
if err := logSpool.Enqueue(validSpoolLogBatch(14, 14)); err != nil {
|
|
t.Fatalf("enqueue later batch: %v", err)
|
|
}
|
|
client := &recoveringSequenceGapLogBatchClient{latestSeq: 10}
|
|
flushed, err := logSpool.Flush(context.Background(), client)
|
|
if err != nil || flushed != 2 {
|
|
t.Fatalf("recover sequence gap: flushed=%d err=%v", flushed, err)
|
|
}
|
|
pending, err := logSpool.Pending()
|
|
if err != nil || len(pending) != 0 {
|
|
t.Fatalf("expected unresendable segments rejected, pending=%+v err=%v", pending, err)
|
|
}
|
|
rejected, err := os.ReadDir(filepath.Join(root, "logs-rejected"))
|
|
if err != nil || len(rejected) != 2 {
|
|
t.Fatalf("expected two rejected segments, rejected=%+v err=%v", rejected, err)
|
|
}
|
|
checksum := func(entries []protocol.LogEntry) (string, error) { return fmt.Sprintf("sha256:%d", len(entries)), nil }
|
|
next := validSpoolLogBatch(0, 0)
|
|
sequence, appended, err := logSpool.EnqueueNextAggregated(context.Background(), next, nil, nil, checksum)
|
|
if err != nil || !appended || sequence != 11 {
|
|
t.Fatalf("expected allocation to resume after platform latest, sequence=%d appended=%t err=%v", sequence, appended, err)
|
|
}
|
|
}
|
|
|
|
func TestLogSpoolDoesNotExtendAcknowledgedSegment(t *testing.T) {
|
|
root := t.TempDir()
|
|
logSpool, err := NewLogSpool(root)
|
|
if err != nil {
|
|
t.Fatalf("new log spool: %v", err)
|
|
}
|
|
checksum := func(entries []protocol.LogEntry) (string, error) { return fmt.Sprintf("sha256:%d", len(entries)), nil }
|
|
if err := logSpool.EnqueueAggregated(validSpoolLogBatch(1, 1), checksum); err != nil {
|
|
t.Fatalf("enqueue acknowledged segment: %v", err)
|
|
}
|
|
logSpool.mu.Lock()
|
|
watermark := logSpool.watermarks["log-1"]
|
|
watermark.Acknowledged = 1
|
|
logSpool.watermarks["log-1"] = watermark
|
|
logSpool.mu.Unlock()
|
|
if err := logSpool.EnqueueAggregated(validSpoolLogBatch(2, 2), checksum); err != nil {
|
|
t.Fatalf("enqueue after acknowledged segment: %v", err)
|
|
}
|
|
entries, err := os.ReadDir(filepath.Join(root, "logs"))
|
|
if err != nil || len(entries) != 2 {
|
|
t.Fatalf("expected the acknowledged segment untouched and a new segment, entries=%+v err=%v", entries, err)
|
|
}
|
|
pending, err := logSpool.Pending()
|
|
if err != nil || len(pending) != 2 || pending[0].FirstSeq != 1 || pending[0].LastSeq != 1 || pending[1].FirstSeq != 2 || pending[1].LastSeq != 2 {
|
|
t.Fatalf("acknowledged segment was extended: pending=%+v err=%v", pending, err)
|
|
}
|
|
}
|
|
|
|
func TestLogSpoolRestoresPendingAllocationWithoutWatermark(t *testing.T) {
|
|
root := t.TempDir()
|
|
first, err := NewLogSpool(root)
|
|
if err != nil {
|
|
t.Fatalf("new first spool: %v", err)
|
|
}
|
|
batch := validSpoolLogBatch(9, 9)
|
|
batch.LogStreamID = "run.endpoint.server.stdout"
|
|
if err := first.Enqueue(batch); err != nil {
|
|
t.Fatalf("enqueue pending batch: %v", err)
|
|
}
|
|
if err := os.Remove(filepath.Join(root, "log-watermarks.json")); err != nil {
|
|
t.Fatalf("remove watermark state: %v", err)
|
|
}
|
|
restarted, err := NewLogSpool(root)
|
|
if err != nil {
|
|
t.Fatalf("restart spool: %v", err)
|
|
}
|
|
called := false
|
|
sequence, err := restarted.NextSequence(context.Background(), batch.LogStreamID, func(context.Context, string) (uint64, error) {
|
|
called = true
|
|
return 3, nil
|
|
})
|
|
if err != nil {
|
|
t.Fatalf("allocate after restart: %v", err)
|
|
}
|
|
if called || sequence != 10 {
|
|
t.Fatalf("expected pending watermark to allocate 10 without remote recovery, got sequence=%d remoteCalled=%t", sequence, called)
|
|
}
|
|
}
|
|
|
|
func TestLogSpoolStartupDoesNotDecodeCorruptPendingSegment(t *testing.T) {
|
|
root := t.TempDir()
|
|
logDir := filepath.Join(root, "logs")
|
|
if err := os.MkdirAll(logDir, 0o755); err != nil {
|
|
t.Fatalf("create log dir: %v", err)
|
|
}
|
|
if err := os.WriteFile(filepath.Join(logDir, "bad-00000000000000000001-00000000000000000001.json"), []byte("{"), 0o600); err != nil {
|
|
t.Fatalf("write corrupt segment: %v", err)
|
|
}
|
|
logSpool, err := NewLogSpool(root)
|
|
if err != nil {
|
|
t.Fatalf("startup should not decode corrupt segment: %v", err)
|
|
}
|
|
pending, err := logSpool.Pending()
|
|
if err != nil || len(pending) != 0 {
|
|
t.Fatalf("expected corrupt segment to be quarantined when read, pending=%+v err=%v", pending, err)
|
|
}
|
|
rejected, err := os.ReadDir(filepath.Join(root, "logs-rejected"))
|
|
if err != nil || len(rejected) != 1 {
|
|
t.Fatalf("expected one rejected corrupt segment, rejected=%+v err=%v", rejected, err)
|
|
}
|
|
}
|
|
|
|
func TestLogSpoolStartupQuarantinesOversizedPendingSegment(t *testing.T) {
|
|
root := t.TempDir()
|
|
logDir := filepath.Join(root, "logs")
|
|
if err := os.MkdirAll(logDir, 0o755); err != nil {
|
|
t.Fatalf("create log dir: %v", err)
|
|
}
|
|
oversized := make([]byte, maxDurableLogSegmentBytes+1)
|
|
if err := os.WriteFile(filepath.Join(logDir, "huge-00000000000000000001-00000000000000000001.json"), oversized, 0o600); err != nil {
|
|
t.Fatalf("write oversized segment: %v", err)
|
|
}
|
|
logSpool, err := NewLogSpool(root)
|
|
if err != nil {
|
|
t.Fatalf("startup should quarantine oversized segment: %v", err)
|
|
}
|
|
pending, err := logSpool.Pending()
|
|
if err != nil || len(pending) != 0 {
|
|
t.Fatalf("expected oversized segment to be quarantined, pending=%+v err=%v", pending, err)
|
|
}
|
|
rejected, err := os.ReadDir(filepath.Join(root, "logs-rejected"))
|
|
if err != nil || len(rejected) != 1 {
|
|
t.Fatalf("expected one rejected oversized segment, rejected=%+v err=%v", rejected, err)
|
|
}
|
|
}
|
|
|
|
func TestLogSpoolSourceCursorDeduplicatesAcknowledgedReplayAfterRestart(t *testing.T) {
|
|
root := t.TempDir()
|
|
checksum := func(entries []protocol.LogEntry) (string, error) {
|
|
return fmt.Sprintf("sha256:%d:%s", len(entries), entries[len(entries)-1].Line), nil
|
|
}
|
|
first, err := NewLogSpool(root)
|
|
if err != nil {
|
|
t.Fatalf("new log spool: %v", err)
|
|
}
|
|
batch := validSpoolLogBatch(0, 0)
|
|
batch.FirstSeq = 0
|
|
batch.LastSeq = 0
|
|
batch.Entries = []protocol.LogEntry{{Timestamp: time.Now().UTC(), Line: "first line"}}
|
|
sequence, appended, err := first.EnqueueNextAggregated(context.Background(), batch, &LogSourceCursor{StartOffset: 0, EndOffset: 11}, nil, checksum)
|
|
if err != nil || !appended || sequence != 1 {
|
|
t.Fatalf("append first cursor: sequence=%d appended=%t err=%v", sequence, appended, err)
|
|
}
|
|
if err := first.Ack(protocol.LogBatchIngestResponse{LogStreamID: batch.LogStreamID, AcceptedFrom: 1, AcceptedTo: 1}); err != nil {
|
|
t.Fatalf("ack first cursor: %v", err)
|
|
}
|
|
restarted, err := NewLogSpool(root)
|
|
if err != nil {
|
|
t.Fatalf("restart log spool: %v", err)
|
|
}
|
|
sequence, appended, err = restarted.EnqueueNextAggregated(context.Background(), batch, &LogSourceCursor{StartOffset: 0, EndOffset: 11}, nil, checksum)
|
|
if err != nil || appended || sequence != 1 {
|
|
t.Fatalf("deduplicate acknowledged cursor: sequence=%d appended=%t err=%v", sequence, appended, err)
|
|
}
|
|
batch.Entries[0].Line = "second line"
|
|
sequence, appended, err = restarted.EnqueueNextAggregated(context.Background(), batch, &LogSourceCursor{StartOffset: 11, EndOffset: 23}, nil, checksum)
|
|
if err != nil || !appended || sequence != 2 {
|
|
t.Fatalf("append next cursor: sequence=%d appended=%t err=%v", sequence, appended, err)
|
|
}
|
|
pending, err := restarted.Pending()
|
|
if err != nil || len(pending) != 1 || pending[0].FirstSeq != 2 || pending[0].Entries[0].Line != "second line" {
|
|
t.Fatalf("unexpected pending cursor batches: pending=%+v err=%v", pending, err)
|
|
}
|
|
}
|
|
|
|
func TestLogSpoolAggregatedSegmentIsReplacedInPlace(t *testing.T) {
|
|
root := t.TempDir()
|
|
logSpool, err := NewLogSpool(root)
|
|
if err != nil {
|
|
t.Fatalf("new log spool: %v", err)
|
|
}
|
|
checksum := func(entries []protocol.LogEntry) (string, error) { return fmt.Sprintf("sha256:%d", len(entries)), nil }
|
|
if err := logSpool.EnqueueAggregated(validSpoolLogBatch(1, 1), checksum); err != nil {
|
|
t.Fatalf("enqueue first segment: %v", err)
|
|
}
|
|
before, err := os.ReadDir(filepath.Join(root, "logs"))
|
|
if err != nil || len(before) != 1 {
|
|
t.Fatalf("read first segment: entries=%+v err=%v", before, err)
|
|
}
|
|
if err := logSpool.EnqueueAggregated(validSpoolLogBatch(2, 2), checksum); err != nil {
|
|
t.Fatalf("aggregate second segment: %v", err)
|
|
}
|
|
after, err := os.ReadDir(filepath.Join(root, "logs"))
|
|
if err != nil || len(after) != 1 || after[0].Name() != before[0].Name() {
|
|
t.Fatalf("aggregation did not replace one stable path: before=%+v after=%+v err=%v", before, after, err)
|
|
}
|
|
pending, err := logSpool.Pending()
|
|
if err != nil || len(pending) != 1 || pending[0].FirstSeq != 1 || pending[0].LastSeq != 2 {
|
|
t.Fatalf("unexpected aggregate after replacement: pending=%+v err=%v", pending, err)
|
|
}
|
|
}
|
|
|
|
func TestLogSpoolDoesNotExtendInflightAggregate(t *testing.T) {
|
|
logSpool, err := NewLogSpool(t.TempDir())
|
|
if err != nil {
|
|
t.Fatalf("new log spool: %v", err)
|
|
}
|
|
checksum := func(entries []protocol.LogEntry) (string, error) { return fmt.Sprintf("sha256:%d", len(entries)), nil }
|
|
if err := logSpool.EnqueueAggregated(validSpoolLogBatch(1, 1), checksum); err != nil {
|
|
t.Fatalf("enqueue first segment: %v", err)
|
|
}
|
|
client := &blockingLogBatchClient{started: make(chan struct{}), release: make(chan struct{})}
|
|
done := make(chan error, 1)
|
|
go func() {
|
|
_, err := logSpool.Flush(context.Background(), client)
|
|
done <- err
|
|
}()
|
|
<-client.started
|
|
if err := logSpool.EnqueueAggregated(validSpoolLogBatch(2, 2), checksum); err != nil {
|
|
t.Fatalf("enqueue while first segment is inflight: %v", err)
|
|
}
|
|
close(client.release)
|
|
if err := <-done; err != nil {
|
|
t.Fatalf("flush inflight segment: %v", err)
|
|
}
|
|
pending, err := logSpool.Pending()
|
|
if err != nil || len(pending) != 1 || pending[0].FirstSeq != 2 || pending[0].LastSeq != 2 {
|
|
t.Fatalf("inflight segment was extended or next segment lost: pending=%+v err=%v", pending, err)
|
|
}
|
|
}
|
|
|
|
type blockingLogBatchClient struct {
|
|
once sync.Once
|
|
started chan struct{}
|
|
release chan struct{}
|
|
}
|
|
|
|
func (client *blockingLogBatchClient) IngestLogBatch(_ context.Context, batch protocol.LogBatchIngestRequest) (protocol.LogBatchIngestResponse, error) {
|
|
client.once.Do(func() { close(client.started) })
|
|
<-client.release
|
|
return protocol.LogBatchIngestResponse{Accepted: true, LogStreamID: batch.LogStreamID, AcceptedFrom: batch.FirstSeq, AcceptedTo: batch.LastSeq}, nil
|
|
}
|
|
|
|
type permanentRejectLogBatchClient struct{}
|
|
|
|
func (permanentRejectLogBatchClient) IngestLogBatch(context.Context, protocol.LogBatchIngestRequest) (protocol.LogBatchIngestResponse, error) {
|
|
return protocol.LogBatchIngestResponse{}, PermanentLogBatchRejection("platform_not_found", nil)
|
|
}
|
|
|
|
type recoveringSequenceGapLogBatchClient struct {
|
|
latestSeq uint64
|
|
progressBatches []protocol.LogBatchIngestRequest
|
|
}
|
|
|
|
func (client *recoveringSequenceGapLogBatchClient) IngestLogBatch(context.Context, protocol.LogBatchIngestRequest) (protocol.LogBatchIngestResponse, error) {
|
|
return protocol.LogBatchIngestResponse{}, PermanentLogBatchRejection("platform_sequence_gap", nil)
|
|
}
|
|
|
|
func (client *recoveringSequenceGapLogBatchClient) LogStreamLatestSeq(_ context.Context, batch protocol.LogBatchIngestRequest) (uint64, error) {
|
|
client.progressBatches = append(client.progressBatches, batch)
|
|
return client.latestSeq, nil
|
|
}
|
|
|
|
type recordingLogBatchClient struct {
|
|
batches []protocol.LogBatchIngestRequest
|
|
}
|
|
|
|
func (client *recordingLogBatchClient) IngestLogBatch(_ context.Context, batch protocol.LogBatchIngestRequest) (protocol.LogBatchIngestResponse, error) {
|
|
client.batches = append(client.batches, batch)
|
|
return protocol.LogBatchIngestResponse{Accepted: true, LogStreamID: batch.LogStreamID, AcceptedFrom: batch.FirstSeq, AcceptedTo: batch.LastSeq}, nil
|
|
}
|
|
|
|
func validSpoolLogBatch(firstSeq uint64, lastSeq uint64) protocol.LogBatchIngestRequest {
|
|
entries := make([]protocol.LogEntry, 0, lastSeq-firstSeq+1)
|
|
for seq := firstSeq; seq <= lastSeq; seq++ {
|
|
entries = append(entries, protocol.LogEntry{Seq: seq, Timestamp: time.Date(2026, 7, 3, 12, 0, int(seq), 0, time.UTC), Level: "info", Line: "line"})
|
|
}
|
|
return protocol.LogBatchIngestRequest{
|
|
RunEndpointID: "run-local",
|
|
SessionToken: "session-token",
|
|
LogStreamID: "log-1",
|
|
ServerInstanceID: "server-1",
|
|
StreamKey: "stdout",
|
|
Source: "process",
|
|
FirstSeq: firstSeq,
|
|
LastSeq: lastSeq,
|
|
Compression: "none",
|
|
Checksum: "sha256:test",
|
|
Entries: entries,
|
|
}
|
|
}
|