package pathway import ( "encoding/json" "errors" "os" "path/filepath" "strings" "testing" ) // persistor_test covers the corruption-recovery contract per // Sprint 2 row 7: malformed JSONL lines must not halt replay. func TestPersistor_MissingFileIsNotError(t *testing.T) { dir := t.TempDir() path := filepath.Join(dir, "nonexistent.jsonl") p, err := NewPersistor(path) if err != nil { t.Fatalf("NewPersistor on missing file should not error, got %v", err) } n, err := p.Replay(func(event) error { return nil }) if err != nil { t.Errorf("Replay on missing file should be 0,nil; got %d, %v", n, err) } if n != 0 { t.Errorf("Replay on missing file replayed %d events, want 0", n) } } func TestPersistor_AppendThenReplay(t *testing.T) { p := mustPersistor(t) if err := p.Append(event{Op: opAdd, Trace: &Trace{UID: "A", Content: json.RawMessage(`{}`)}}); err != nil { t.Fatalf("Append: %v", err) } if err := p.Append(event{Op: opAdd, Trace: &Trace{UID: "B", Content: json.RawMessage(`{}`)}}); err != nil { t.Fatalf("Append: %v", err) } var seen []string n, err := p.Replay(func(e event) error { if e.Trace != nil { seen = append(seen, e.Trace.UID) } return nil }) if err != nil { t.Fatalf("Replay: %v", err) } if n != 2 { t.Errorf("Replay applied %d events, want 2", n) } if len(seen) != 2 || seen[0] != "A" || seen[1] != "B" { t.Errorf("seen = %v, want [A B]", seen) } } func TestPersistor_CorruptedLines_Skipped(t *testing.T) { p := mustPersistor(t) // Mix of valid and corrupted lines. good1 := mustMarshal(t, event{Op: opAdd, Trace: &Trace{UID: "A", Content: json.RawMessage(`{}`)}}) bad := []byte(`{this is not json}`) good2 := mustMarshal(t, event{Op: opAdd, Trace: &Trace{UID: "B", Content: json.RawMessage(`{}`)}}) emptyLine := []byte(``) good3 := mustMarshal(t, event{Op: opAdd, Trace: &Trace{UID: "C", Content: json.RawMessage(`{}`)}}) contents := []byte{} for _, line := range [][]byte{good1, bad, good2, emptyLine, good3} { contents = append(contents, line...) contents = append(contents, '\n') } if err := os.WriteFile(p.Path(), contents, 0o644); err != nil { t.Fatalf("write file: %v", err) } var applied []string n, err := p.Replay(func(e event) error { if e.Trace != nil { applied = append(applied, e.Trace.UID) } return nil }) if err != nil { t.Fatalf("Replay: %v", err) } // 3 valid + 1 bad + 1 empty (skipped silently) = 3 applied. if n != 3 { t.Errorf("Replay applied %d, want 3 (1 corrupt line skipped)", n) } if len(applied) != 3 || applied[0] != "A" || applied[1] != "B" || applied[2] != "C" { t.Errorf("applied = %v, want [A B C]", applied) } } func TestPersistor_ApplyError_Skipped(t *testing.T) { // If the apply function returns error for an event, replay // should keep going (the error is logged, not raised). p := mustPersistor(t) _ = p.Append(event{Op: opAdd, Trace: &Trace{UID: "A", Content: json.RawMessage(`{}`)}}) _ = p.Append(event{Op: opAdd, Trace: &Trace{UID: "B", Content: json.RawMessage(`{}`)}}) _ = p.Append(event{Op: opAdd, Trace: &Trace{UID: "C", Content: json.RawMessage(`{}`)}}) count := 0 n, err := p.Replay(func(e event) error { if e.Trace != nil && e.Trace.UID == "B" { return errors.New("simulated apply error on B") } count++ return nil }) if err != nil { t.Fatalf("Replay: %v", err) } if n != 2 || count != 2 { t.Errorf("Replay applied %d (callback called %d), want 2 each (B's error skipped)", n, count) } } func TestPersistor_NewPersistor_EmptyPath_Errors(t *testing.T) { _, err := NewPersistor("") if err == nil { t.Error("NewPersistor with empty path should error") } } func TestPersistor_CreatesParentDir(t *testing.T) { dir := t.TempDir() nested := filepath.Join(dir, "nested", "deep", "pathway.jsonl") p, err := NewPersistor(nested) if err != nil { t.Fatalf("NewPersistor: %v", err) } if err := p.Append(event{Op: opAdd, Trace: &Trace{UID: "A", Content: json.RawMessage(`{}`)}}); err != nil { t.Fatalf("Append after creating nested dir: %v", err) } } func TestPersistor_LongLine_HandlesUpTo1MiB(t *testing.T) { p := mustPersistor(t) // Build a content blob ~750 KiB so the JSON line is ~800 KiB // (under the 1 MiB scanner cap). blob := strings.Repeat("x", 750*1024) bigContent, _ := json.Marshal(map[string]string{"data": blob}) tr := &Trace{UID: "BIG", Content: bigContent} if err := p.Append(event{Op: opAdd, Trace: tr}); err != nil { t.Fatalf("Append big trace: %v", err) } count := 0 n, _ := p.Replay(func(e event) error { if e.Trace != nil && e.Trace.UID == "BIG" { count++ } return nil }) if n != 1 || count != 1 { t.Errorf("big-line replay: got %d events / %d matches, want 1 each", n, count) } } // ── helpers ── func mustPersistor(t *testing.T) *Persistor { t.Helper() dir := t.TempDir() path := filepath.Join(dir, "test.jsonl") p, err := NewPersistor(path) if err != nil { t.Fatalf("NewPersistor: %v", err) } return p } func mustMarshal(t *testing.T, e event) []byte { t.Helper() b, err := json.Marshal(e) if err != nil { t.Fatalf("marshal: %v", err) } return b }