package logstorage import ( "bytes" "os" "testing" "time" ) func TestClearPortCompactsOnlyTargetAndPreservesSequences(t *testing.T) { root := t.TempDir() s := openTestStore(t, Options{Root: root}) now := time.Now().UTC() for _, channel := range []string{"CH01", "CH02", "CH01"} { if _, err := s.Append(ClassPort, now, map[string]string{"channel": channel, "message": "private-" + channel}); err != nil { t.Fatal(err) } } for _, class := range []string{ClassAudit, ClassSystem, ClassLink} { if _, err := s.Append(class, now, map[string]string{"channel": "CH01", "message": "protected"}); err != nil { t.Fatal(err) } } if err := s.ApplyRetention(Retention{365, 365, 365}); err != nil { t.Fatal(err) } // closed segment if _, err := s.Append(ClassPort, now, map[string]string{"channel": "CH01", "message": "active"}); err != nil { t.Fatal(err) } if count, err := s.ClearPort("CH01"); err != nil || count != 3 { t.Fatalf("clear = %d, %v", count, err) } paths, _ := s.closedSegmentsLocked(ClassPort) paths = append(paths, s.activePath(ClassPort)) for _, path := range paths { data, err := os.ReadFile(path) if err != nil || bytes.Contains(data, []byte("CH01")) { t.Fatalf("target remains on disk: %s, %v", path, err) } } reopened := openTestStore(t, Options{Root: root}) page, err := reopened.Page(ClassPort, 0, 500) if err != nil || len(page) != 1 || page[0].Sequence != 2 { t.Fatalf("remaining page = %+v, %v", page, err) } for _, class := range []string{ClassAudit, ClassSystem, ClassLink} { rows, err := reopened.Read(class) if err != nil || len(rows) != 1 { t.Fatalf("protected %s logs changed: %d, %v", class, len(rows), err) } } next, err := reopened.Append(ClassPort, now, map[string]string{"channel": "CH01", "message": "new"}) if err != nil || next.Sequence != 5 { t.Fatalf("new append = %+v, %v", next, err) } var exported bytes.Buffer if err := reopened.Export(ClassPort, &exported); err != nil || bytes.Contains(exported.Bytes(), []byte("private-CH01")) { t.Fatalf("export leaked cleared records: %v", err) } if n, err := reopened.ClearPort("CH01"); err != nil || n != 1 { t.Fatalf("second clear = %d, %v", n, err) } } func TestClearPortCheckpointRecoversInterruptedCompaction(t *testing.T) { root := t.TempDir() s := openTestStore(t, Options{Root: root}) for _, channel := range []string{"CH01", "CH02"} { if _, err := s.Append(ClassPort, time.Now(), map[string]string{"channel": channel}); err != nil { t.Fatal(err) } } original, err := os.ReadFile(s.activePath(ClassPort)) if err != nil { t.Fatal(err) } if _, err := s.ClearPort("CH01"); err != nil { t.Fatal(err) } // Simulate a crash after the durable clear checkpoint but before rewriting // this one segment. Recovery must finish compaction, not resurrect records. if err := os.WriteFile(s.activePath(ClassPort), original, 0o600); err != nil { t.Fatal(err) } reopened := openTestStore(t, Options{Root: root}) rows, err := reopened.Read(ClassPort) if err != nil || len(rows) != 1 || bytes.Contains(rows[0], []byte("CH01")) { t.Fatalf("recovered records = %s, %v", rows, err) } if n, err := reopened.ClearPort("CH02"); err != nil || n != 1 { t.Fatalf("clear all remaining = %d, %v", n, err) } reopened = openTestStore(t, Options{Root: root}) row, err := reopened.Append(ClassPort, time.Now(), map[string]string{"channel": "CH02"}) if err != nil || row.Sequence != 3 { t.Fatalf("sequence reused after clearing all: %+v, %v", row, err) } } func TestClearPortRejectsBroadOrInvalidTargets(t *testing.T) { s := openTestStore(t, Options{}) for _, channel := range []string{"", "all", "CH00", "CH49", "CH1", "../audit", "CH01 "} { if _, err := s.ClearPort(channel); err == nil { t.Errorf("accepted %q", channel) } } if s.PortClearRevision() != 0 { t.Fatal("invalid requests changed deletion state") } }