package main import ( "bytes" "log/slog" "strings" "testing" "time" "github.com/pingular/pingularity/internal/config" "github.com/pingular/pingularity/internal/settings" "github.com/pingular/pingularity/internal/stats" ) // A BIG CLEANUP IS A PASS THAT CAN BE INTERRUPTED. // // Prune deletes in chunks with a wait after every full one, so a pass over a // big backlog lasts seconds or minutes where it used to be one statement. Two // things now land inside a pass that used to land beside it: a shutdown, and // another writer holding the database. Neither is a failure. What went stays // gone, the rest goes at the next pass, and the log says so in those words. // seedExpiredSamples writes n latency rows a second apart, all older than any // window the tests set, in one statement on the second connection. func (f *prunerStore) seedExpiredSamples(n int) { f.t.Helper() newest := time.Now().Add(-400 * 24 * time.Hour).Unix() if _, err := f.db2.ExecContext(f.ctx, ` WITH RECURSIVE n(i) AS (SELECT 0 UNION ALL SELECT i + 1 FROM n WHERE i < ? - 1) INSERT INTO samples (ts, target, latency_ms, success, family) SELECT ? - i, 'a', 10.0, 1, 'ipv4' FROM n`, n, newest); err != nil { f.t.Fatalf("seed %d samples: %v", n, err) } if got := f.count("samples"); got != int64(n) { f.t.Fatalf("seed: %d samples, want %d", got, n) } } // tenDayController stores a ten day latency window, keeps the rest forever, // and boots settings on it the way run() does. func tenDayController(t *testing.T, f *prunerStore) *settings.Controller { t.Helper() if err := f.st.SetSettings(f.ctx, map[string]string{ "retention_s": "864000", "speed_retention_s": "0", "downtime_retention_s": "0", }); err != nil { t.Fatal(err) } set, err := settings.New(f.ctx, f.st, defaultSettings(config.Default())) if err != nil { t.Fatal(err) } if !set.Loaded() { t.Fatal("fixture: the controller must be loaded") } return set } // A shutdown in the middle of a big pass. The pruner has to let go inside the // shutdown budget, leave the rows it had not reached, and report a stop rather // than an error. The next start finishes the job. func TestPrunerStopsMidPassAndFinishesLater(t *testing.T) { f := openPrunerStore(t) const seeded = 100000 // five full chunks, so most of a second in waits alone f.seedExpiredSamples(seeded) set := tenDayController(t, f) passes := prunePasses() var logs bytes.Buffer p := &program{store: f.st, log: slog.New(slog.NewTextHandler(&logs, nil))} stop, _ := runPrunerFor(t, p, set) defer stop() if !waitFor(func() bool { return f.count("samples") < seeded }, 5*time.Second) { t.Fatal("no chunk was deleted within 5s of the grace") } began := time.Now() stop() // fails the test itself if the pruner takes more than 5s to return took := time.Since(began) left := f.count("samples") if left == 0 { t.Fatalf("the pass finished before it could be stopped; %d rows is too small a fixture for this machine", seeded) } if took > shutdownWorkerGrace { t.Errorf("the pruner took %v to stop in the middle of a pass; workers get %v at shutdown, and a chunk or "+ "a wait has to end inside it", took, shutdownWorkerGrace) } if got := prunePasses(); got != passes { t.Errorf("db.prune_count moved by %d: a stopped pass is not a finished prune", got-passes) } if strings.Contains(logs.String(), "level=ERROR") { t.Errorf("a stop for shutdown was logged as an error:\n%s", logs.String()) } if !strings.Contains(logs.String(), "prune stopped for shutdown") { t.Errorf("the stopped pass was not reported:\n%s", logs.String()) } // The next start picks up where this one stopped. logs.Reset() stop2, _ := runPrunerFor(t, p, set) defer stop2() if !waitFor(func() bool { return prunePasses() > passes }, 30*time.Second) { t.Fatalf("the second run did not finish the pass within 30s; %d rows left", f.count("samples")) } stop2() if got := f.count("samples"); got != 0 { t.Errorf("%d expired rows left after the second run finished its pass, want 0", got) } if strings.Contains(logs.String(), "level=ERROR") { t.Errorf("the second run failed:\n%s", logs.String()) } } // Another writer holds the database for longer than a chunk will wait, which // a restore can do. The pass gives way and says so as a warning, the ticker // stays, and the pass after the writer lets go removes the rows. func TestPrunerGivesWayToAnotherWriter(t *testing.T) { if testing.Short() { t.Skip("waits out the 5s busy timeout") } f := openPrunerStore(t) f.seedExpiredSamples(100) set := tenDayController(t, f) prevTick := pruneInterval pruneInterval = 100 * time.Millisecond t.Cleanup(func() { pruneInterval = prevTick }) // One connection, so the transaction that takes the writer is the one // that gives it back. holder, err := f.db2.Conn(f.ctx) if err != nil { t.Fatal(err) } defer holder.Close() if _, err := holder.ExecContext(f.ctx, `BEGIN IMMEDIATE`); err != nil { t.Fatal(err) } held := true defer func() { if held { holder.ExecContext(f.ctx, `ROLLBACK`) //nolint:errcheck // only the failure path } }() passes := prunePasses() busy := stats.Lifetime().Counters["db.busy"] var logs bytes.Buffer p := &program{store: f.st, log: warnOnlyLog(&logs)} stop, running := runPrunerFor(t, p, set) defer stop() if !waitFor(func() bool { return stats.Lifetime().Counters["db.busy"] > busy }, 15*time.Second) { t.Fatal("no chunk gave up on the held writer within 15s") } if got := f.count("samples"); got != 100 { t.Errorf("%d samples left while the writer was held, want all 100", got) } if !running() { t.Fatal("runPruner exited after the pass that gave way; nothing would prune once the writer is free") } if _, err := holder.ExecContext(f.ctx, `ROLLBACK`); err != nil { t.Fatal(err) } held = false if !waitFor(func() bool { return prunePasses() > passes }, 15*time.Second) { t.Fatal("no pass finished within 15s of the writer letting go") } stop() if got := f.count("samples"); got != 0 { t.Errorf("%d expired rows left after the writer let go, want 0", got) } if !strings.Contains(logs.String(), "level=WARN") || !strings.Contains(logs.String(), "prune gave way to another writer") { t.Errorf("the pass that gave way must be reported as a warning:\n%s", logs.String()) } if strings.Contains(logs.String(), "level=ERROR") { t.Errorf("giving way to another writer was logged as an error:\n%s", logs.String()) } } // A backup is being restored when the hourly pass comes due. The pass is // skipped, not failed: nothing is deleted, the log says why without a warning // or an error, the ticker stays, and the pass after the restore has let go // removes the rows. func TestPrunerSkipsAPassThatComesDueDuringARestore(t *testing.T) { f := openPrunerStore(t) f.seedExpiredSamples(100) set := tenDayController(t, f) prevTick := pruneInterval pruneInterval = 100 * time.Millisecond t.Cleanup(func() { pruneInterval = prevTick }) release := f.st.HoldCleanup() // what a restore takes before its first row defer release() passes := prunePasses() skipped := func() int64 { return stats.Lifetime().Counters["db.prune_skipped_restore"] } before := skipped() var logs bytes.Buffer p := &program{store: f.st, log: slog.New(slog.NewTextHandler(&logs, nil))} stop, running := runPrunerFor(t, p, set) defer stop() if !waitFor(func() bool { return skipped() > before }, 15*time.Second) { t.Fatal("no pass was counted as skipped for the restore within 15s") } if got := f.count("samples"); got != 100 { t.Errorf("%d samples left while the restore held the cleanup off, want all 100", got) } if got := prunePasses(); got != passes { t.Errorf("db.prune_count moved by %d: a skipped pass is not a finished prune", got-passes) } if !running() { t.Fatal("runPruner exited after the skipped pass; nothing would prune once the restore is done") } release() if !waitFor(func() bool { return prunePasses() > passes }, 15*time.Second) { t.Fatal("no pass finished within 15s of the restore letting go") } stop() if got := f.count("samples"); got != 0 { t.Errorf("%d expired rows left after the restore let go, want 0", got) } if !strings.Contains(logs.String(), "prune skipped while a backup is being restored") { t.Errorf("the skipped pass was not reported:\n%s", logs.String()) } if strings.Contains(logs.String(), "level=ERROR") || strings.Contains(logs.String(), "level=WARN") { t.Errorf("a pass skipped for a restore was logged as a fault:\n%s", logs.String()) } } // A window raised while a big pass runs takes effect at the pass's next // chunk, not at the pass after: runPruner hands the pass the controller's // windows to read as it goes, not a copy taken as it started. Here the stored // latency window is ten days over two hundred thousand readings twenty days // old, and it is raised to thirty days on the controller once the pass has // deleted some. Every chunk that starts after the raise reads it, so the pass // deletes at most the one chunk under way as the raise lands, and the rest, // which thirty days keep, stays. func TestPrunerFollowsAWindowRaisedMidPass(t *testing.T) { f := openPrunerStore(t) const seeded = 200000 // ten full chunks, a second and a half in waits alone newest := time.Now().Add(-20 * 24 * time.Hour).Unix() if _, err := f.db2.ExecContext(f.ctx, ` WITH RECURSIVE n(i) AS (SELECT 0 UNION ALL SELECT i + 1 FROM n WHERE i < ? - 1) INSERT INTO samples (ts, target, latency_ms, success, family) SELECT ? - i, 'a', 10.0, 1, 'ipv4' FROM n`, seeded, newest); err != nil { t.Fatalf("seed %d samples: %v", seeded, err) } set := tenDayController(t, f) passes := prunePasses() var logs bytes.Buffer p := &program{store: f.st, log: slog.New(slog.NewTextHandler(&logs, nil))} stop, _ := runPrunerFor(t, p, set) defer stop() if !waitFor(func() bool { return f.count("samples") < seeded }, 10*time.Second) { t.Fatal("no chunk was deleted within 10s of the grace") } thirty := 30 * 24 * time.Hour if _, err := set.Update(f.ctx, settings.Patch{Retention: &thirty}); err != nil { t.Fatalf("raise the latency window: %v", err) } // Counted once the raise is in force: a chunk that starts from here on // reads it, and one already under way is the only one that does not. const chunk = 20000 // store.pruneChunkRows raisedAt := f.count("samples") if raisedAt <= chunk { t.Fatalf("the pass had deleted all but %d of %d readings by the time the window was raised; %d rows is too "+ "small a fixture for this machine", raisedAt, seeded, seeded) } if !waitFor(func() bool { return prunePasses() > passes }, 30*time.Second) { t.Fatalf("the pass did not finish within 30s; %d rows left", f.count("samples")) } stop() if left := f.count("samples"); left < raisedAt-chunk { t.Errorf("%d readings, twenty days old, were left as the window was raised to thirty days and %d after the "+ "pass: it went on past the chunk under way, at the ten days it started with", raisedAt, left) } if strings.Contains(logs.String(), "level=ERROR") { t.Errorf("the pass failed:\n%s", logs.String()) } }