Handle malformed events during JetStream replay with proper error reporting
All checks were successful
CI / build (pull_request) Successful in 17s
All checks were successful
CI / build (pull_request) Successful in 17s
Add ReplayError and ReplayResult types to capture information about malformed events encountered during replay. This allows callers to inspect and handle corrupted data rather than having it silently skipped. Key changes: - Add ReplayError type with sequence number, raw data, and underlying error - Add ReplayResult type containing both successfully parsed events and errors - Add EventStoreWithErrors interface for stores that can report replay errors - Implement GetEventsWithErrors on JetStreamEventStore - Update GetEvents to maintain backward compatibility (still skips malformed) - Add comprehensive unit tests for the new types This addresses the issue of silent data loss during event-sourced replay by giving callers visibility into data quality issues. Closes #39 Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
127
event_test.go
127
event_test.go
@@ -1208,3 +1208,130 @@ func TestEvent_MetadataAllHelpersRoundTrip(t *testing.T) {
|
||||
t.Errorf("GetSpanID mismatch: got %q", decoded.GetSpanID())
|
||||
}
|
||||
}
|
||||
|
||||
// Tests for ReplayError and ReplayResult types
|
||||
|
||||
func TestReplayError_Error(t *testing.T) {
|
||||
err := &ReplayError{
|
||||
SequenceNumber: 42,
|
||||
RawData: []byte(`invalid json`),
|
||||
Err: json.Unmarshal([]byte(`{`), &struct{}{}),
|
||||
}
|
||||
|
||||
errMsg := err.Error()
|
||||
if !strings.Contains(errMsg, "42") {
|
||||
t.Errorf("expected error message to contain sequence number, got: %s", errMsg)
|
||||
}
|
||||
if !strings.Contains(errMsg, "unmarshal") || !strings.Contains(errMsg, "failed") {
|
||||
t.Errorf("expected error message to contain 'failed' and 'unmarshal', got: %s", errMsg)
|
||||
}
|
||||
}
|
||||
|
||||
func TestReplayError_Unwrap(t *testing.T) {
|
||||
innerErr := json.Unmarshal([]byte(`{`), &struct{}{})
|
||||
err := &ReplayError{
|
||||
SequenceNumber: 1,
|
||||
RawData: []byte(`{`),
|
||||
Err: innerErr,
|
||||
}
|
||||
|
||||
unwrapped := err.Unwrap()
|
||||
if unwrapped != innerErr {
|
||||
t.Errorf("expected Unwrap to return inner error")
|
||||
}
|
||||
}
|
||||
|
||||
func TestReplayResult_HasErrors(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
result *ReplayResult
|
||||
expected bool
|
||||
}{
|
||||
{
|
||||
name: "no errors",
|
||||
result: &ReplayResult{Events: []*Event{}, Errors: []ReplayError{}},
|
||||
expected: false,
|
||||
},
|
||||
{
|
||||
name: "nil errors slice",
|
||||
result: &ReplayResult{Events: []*Event{}, Errors: nil},
|
||||
expected: false,
|
||||
},
|
||||
{
|
||||
name: "has errors",
|
||||
result: &ReplayResult{
|
||||
Events: []*Event{},
|
||||
Errors: []ReplayError{
|
||||
{SequenceNumber: 1, RawData: []byte(`bad`), Err: nil},
|
||||
},
|
||||
},
|
||||
expected: true,
|
||||
},
|
||||
{
|
||||
name: "has events and errors",
|
||||
result: &ReplayResult{
|
||||
Events: []*Event{{ID: "evt-1"}},
|
||||
Errors: []ReplayError{
|
||||
{SequenceNumber: 2, RawData: []byte(`bad`), Err: nil},
|
||||
},
|
||||
},
|
||||
expected: true,
|
||||
},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
if got := tt.result.HasErrors(); got != tt.expected {
|
||||
t.Errorf("HasErrors() = %v, want %v", got, tt.expected)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestReplayResult_EmptyResult(t *testing.T) {
|
||||
result := &ReplayResult{
|
||||
Events: []*Event{},
|
||||
Errors: []ReplayError{},
|
||||
}
|
||||
|
||||
if result.HasErrors() {
|
||||
t.Error("expected HasErrors() to return false for empty result")
|
||||
}
|
||||
if len(result.Events) != 0 {
|
||||
t.Errorf("expected 0 events, got %d", len(result.Events))
|
||||
}
|
||||
}
|
||||
|
||||
func TestReplayError_WithZeroSequence(t *testing.T) {
|
||||
err := &ReplayError{
|
||||
SequenceNumber: 0,
|
||||
RawData: []byte(`corrupted`),
|
||||
Err: json.Unmarshal([]byte(`not-json`), &struct{}{}),
|
||||
}
|
||||
|
||||
errMsg := err.Error()
|
||||
if !strings.Contains(errMsg, "sequence 0") {
|
||||
t.Errorf("expected error message to contain 'sequence 0', got: %s", errMsg)
|
||||
}
|
||||
}
|
||||
|
||||
func TestReplayError_WithLargeRawData(t *testing.T) {
|
||||
largeData := make([]byte, 1024*1024) // 1MB
|
||||
for i := range largeData {
|
||||
largeData[i] = 'x'
|
||||
}
|
||||
|
||||
err := &ReplayError{
|
||||
SequenceNumber: 999,
|
||||
RawData: largeData,
|
||||
Err: json.Unmarshal(largeData, &struct{}{}),
|
||||
}
|
||||
|
||||
// Should be able to create the error without issues
|
||||
if len(err.RawData) != 1024*1024 {
|
||||
t.Errorf("expected RawData to be preserved, got length %d", len(err.RawData))
|
||||
}
|
||||
|
||||
// Error() should still work
|
||||
_ = err.Error()
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user