diff --git a/cmd/partition/main.go b/cmd/partition/main.go index 0c6238f..e1388fb 100644 --- a/cmd/partition/main.go +++ b/cmd/partition/main.go @@ -88,8 +88,14 @@ func run(ctx context.Context, out io.Writer, dsn string, span store.PartitionSpa // against a default nobody chose. dropped, err := store.Retain(ctx, dsn, time.Now(), retain, span) for _, d := range dropped { - fmt.Fprintf(out, "dropped %s covering %s to %s, %d rows\n", - d.Name, d.RangeStart.Format(time.DateOnly), d.RangeEnd.Format(time.DateOnly), d.Rows) + rows := fmt.Sprintf("%d rows", d.Rows) + if !d.Counted { + // Never "0 rows": nobody counted this one, and printing a zero + // would report a measurement that was not taken. + rows = "an uncounted number of rows" + } + fmt.Fprintf(out, "dropped %s covering %s to %s, %s\n", + d.Name, d.RangeStart.Format(time.DateOnly), d.RangeEnd.Format(time.DateOnly), rows) } if err != nil { // Whatever was dropped before the failure is still dropped, and the diff --git a/internal/store/partition.go b/internal/store/partition.go index 795a8c1..3d715be 100644 --- a/internal/store/partition.go +++ b/internal/store/partition.go @@ -41,11 +41,6 @@ type PartitionSpan struct { Behind int } -// WithDefaults is withDefaults for a caller outside this package: the -// retention flag has to know what the horizon actually keeps, and the defaults -// are the answer when the flags were not given. -func (s PartitionSpan) WithDefaults() PartitionSpan { return s.withDefaults() } - func (s PartitionSpan) withDefaults() PartitionSpan { if s.Ahead <= 0 { s.Ahead = 4 diff --git a/internal/store/retention.go b/internal/store/retention.go index 845ce64..5f2e746 100644 --- a/internal/store/retention.go +++ b/internal/store/retention.go @@ -24,6 +24,8 @@ import ( // makes DETACH CONCURRENTLY fail outright and every run after it, // 2. refuse any partition that is not wholly past the boundary, which is the // board's criterion: an active partition cannot be dropped by accident, +// 2a. finish what a run that stopped after a detach left behind, because that +// state is invisible to the search in step 2, // 3. write the record, before anything is touched, which is also where a week // that was dropped once and is back is refused, because the record is the // only thing that remembers the drop: that refusal arrives per candidate, @@ -41,6 +43,11 @@ import ( // next clean run returned nil having seen nothing, because a detached table is // not inherited and nothing looks for it. // +// Writing the record was only half of that. Until step 2a existed the row was +// written and never read, so a run that stopped after the detach left the same +// invisible partition with a note beside it that nothing acted on. The record +// is what the criterion asks for, and a record no run reads does not meet it. +// // The lock is step zero for a reason measured on this branch. Step one reads // pg_get_expr(relpartbound), which takes a lock on every partition and so waits // behind a detach already running. With that read ahead of the lock, a second @@ -92,6 +99,14 @@ type Dropped struct { RangeStart time.Time RangeEnd time.Time Rows int64 + + // Counted is false when nobody ever counted this partition, which happens + // only when a record is resumed for a table that is already gone and whose + // count was never written. Rows is zero then, and zero is a measurement, so + // a caller that prints the number has to print this instead: a partition + // nobody counted and one that held nothing look the same otherwise, and the + // record exists to keep them apart. + Counted bool } // bounds parses what pg_get_expr renders for a range partition. The shape is @@ -198,6 +213,13 @@ func Retain(ctx context.Context, dsn string, now time.Time, keep time.Duration, return nil, err } + // Before any candidate is looked for, because a run that stopped after the + // detach left something no candidate search can see. + out, err := resumeUnfinished(ctx, conn, now) + if err != nil { + return out, err + } + boundary := now.UTC().Add(-keep) candidates, err := pastPartitions(ctx, conn, boundary) if err != nil { @@ -218,7 +240,6 @@ func Retain(ctx context.Context, dsn string, now time.Time, keep time.Duration, candidates = append([]candidate{candidates[i]}, append(candidates[:i:i], candidates[i+1:]...)...) } - var out []Dropped for _, c := range candidates { d, err := dropOne(ctx, conn, c, now) if err != nil { @@ -361,7 +382,104 @@ func dropOne(ctx context.Context, conn *pgx.Conn, c candidate, now time.Time) (D if _, err := conn.Exec(ctx, `UPDATE events_partition_drops SET dropped_at = $1 WHERE id = $2`, now.UTC(), id); err != nil { return Dropped{}, fmt.Errorf("retention: stamp %s: %w", c.name, err) } - return Dropped{Name: c.name, RangeStart: c.from, RangeEnd: c.to, Rows: rows}, nil + return Dropped{Name: c.name, RangeStart: c.from, RangeEnd: c.to, Rows: rows, Counted: true}, nil +} + +// resumeUnfinished finishes what a run that stopped after the detach left +// behind, and it runs before anything else looks for a candidate because that +// state is invisible to every candidate search: a detached partition is not +// inherited, so pastPartitions and pendingDetach look straight past it, and the +// only thing that knows it ever existed is the record written before it was +// touched. +// +// Measured on this branch before it existed: a run stopped between the detach +// and the drop left the partition on disk with its rows, no longer inherited, +// and the next run returned nil having seen nothing. The record was there the +// whole time and nothing read it. This is what turns that record from a trace +// into the thing the criterion says it is. +// +// Reaching that state does not need a kill: cmd/partition runs on +// context.Background() and catches no signal, so Ctrl-C, a scheduler's SIGTERM, +// the OOM killer or a dropped connection all end the process in the same place, +// and since this package sets lock_timeout, so does a lock it could not get. +// +// Two shapes. The partition is still there, which means the drop never ran or +// never finished: count it if the count was never taken, then drop and stamp. +// Or it is gone and only the stamp is missing, which is a run that died between +// the two, and then the row is closed as it stands: a count nobody took stays +// null rather than becoming a zero somebody could read as a measurement. +func resumeUnfinished(ctx context.Context, conn *pgx.Conn, now time.Time) ([]Dropped, error) { + type unfinished struct { + id int64 + name string + from, to time.Time + rows *int64 + counted bool + onDisk bool + } + rows, err := conn.Query(ctx, ` + SELECT d.id, d.partition_name, d.range_start, d.range_end, d.rows_dropped, + d.detached_at IS NOT NULL, c.oid IS NOT NULL + FROM events_partition_drops d + LEFT JOIN pg_class c + ON c.relname = d.partition_name AND c.relnamespace = 'public'::regnamespace + LEFT JOIN pg_inherits i ON i.inhrelid = c.oid + WHERE d.dropped_at IS NULL AND i.inhrelid IS NULL + ORDER BY d.range_start`) + if err != nil { + return nil, fmt.Errorf("retention: look for an unfinished drop: %w", err) + } + var open []unfinished + for rows.Next() { + var u unfinished + if err := rows.Scan(&u.id, &u.name, &u.from, &u.to, &u.rows, &u.counted, &u.onDisk); err != nil { + return nil, fmt.Errorf("retention: %w", err) + } + open = append(open, u) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("retention: %w", err) + } + rows.Close() + + var out []Dropped + for _, u := range open { + var n int64 + counted := true + switch { + case u.onDisk: + n = 0 + if u.counted && u.rows != nil { + n = *u.rows + } else { + if err := conn.QueryRow(ctx, fmt.Sprintf(`SELECT count(*) FROM %s`, quoteIdent(u.name))).Scan(&n); err != nil { + return out, fmt.Errorf("retention: count the unfinished %s: %w", u.name, err) + } + if _, err := conn.Exec(ctx, ` + UPDATE events_partition_drops SET rows_dropped = $1, detached_at = $2 WHERE id = $3`, + n, now.UTC(), u.id); err != nil { + return out, fmt.Errorf("retention: record the count for the unfinished %s: %w", u.name, err) + } + } + if _, err := conn.Exec(ctx, fmt.Sprintf(`DROP TABLE %s`, quoteIdent(u.name))); err != nil { + return out, fmt.Errorf("retention: drop the unfinished %s: %w", u.name, err) + } + case u.rows != nil: + n = *u.rows + default: + // The table is gone and no count was ever written, which a run + // cannot produce on its own: the count goes in before the drop. It + // is what a partition dropped by hand next to an open record looks + // like, and the row is closed as it stands rather than gaining a + // zero nobody measured. + counted = false + } + if _, err := conn.Exec(ctx, `UPDATE events_partition_drops SET dropped_at = $1 WHERE id = $2`, now.UTC(), u.id); err != nil { + return out, fmt.Errorf("retention: stamp the unfinished %s: %w", u.name, err) + } + out = append(out, Dropped{Name: u.name, RangeStart: u.from, RangeEnd: u.to, Rows: n, Counted: counted}) + } + return out, nil } // quoteIdent is the only safe way to put a name this package read from the diff --git a/internal/store/retention_test.go b/internal/store/retention_test.go index bf9e7b3..49dfcd9 100644 --- a/internal/store/retention_test.go +++ b/internal/store/retention_test.go @@ -642,3 +642,211 @@ func TestADetachThatCannotGetItsLockStopsAndTheNextRunFinishesIt(t *testing.T) { t.Fatalf("the record for %s is not stamped, and the drop happened", name) } } + +// A run that gets past the detach and then stops is the state no candidate +// search can see, because a detached partition is not inherited. The record is +// the only thing left that knows, so the next run has to read it. The state is +// produced by stopping a real run rather than by staging it with SQL. +func TestARunStoppedAfterTheDetachIsFinishedByTheNextOne(t *testing.T) { + dsn := withHistory(t) + c := conn(t, dsn) + ctx := context.Background() + old := olderWeek(6) + seed(t, c, old, 5) + name := partitionName(weekStart(old)) + + prev := lockTimeout + lockTimeout = 3 * time.Second + t.Cleanup(func() { lockTimeout = prev }) + + first := conn(t, dsn) + if _, err := first.Exec(ctx, `BEGIN`); err != nil { + t.Fatal(err) + } + if _, err := first.Exec(ctx, fmt.Sprintf(`SELECT count(*) FROM %s`, name)); err != nil { + t.Fatal(err) + } + + // Bounded, so a run that stops waiting instead of stopping fails this test + // rather than holding the package until the ten minute panic. + held, cancel := context.WithTimeout(ctx, 40*time.Second) + defer cancel() + stopped := make(chan error, 1) + go func() { + _, err := Retain(held, dsn, time.Now(), 28*24*time.Hour, PartitionSpan{Behind: horizonBehind}) + stopped <- err + }() + + deadline := time.After(20 * time.Second) + for { + if rowsIn(t, c, `SELECT count(*) FROM pg_inherits i JOIN pg_class p ON p.oid = i.inhrelid + WHERE p.relname = $1 AND i.inhdetachpending`, name) == 1 { + break + } + select { + case <-deadline: + t.Fatal("the detach never reached the pending mark") + case err := <-stopped: + t.Fatalf("the run ended before the detach was pending: %v", err) + case <-time.After(20 * time.Millisecond): + } + } + + // The block goes on the record table, not on the partition. A second reader + // of the partition would hold the detach as well, measured: the detach took + // the lock timeout and stayed pending, which is the state the test before + // this one already covers. Holding events_partition_drops instead lets the + // detach finish and stops the statement after it, which is the state + // nothing could see. + second := conn(t, dsn) + if _, err := second.Exec(ctx, `BEGIN`); err != nil { + t.Fatal(err) + } + if _, err := second.Exec(ctx, `LOCK TABLE events_partition_drops IN ACCESS EXCLUSIVE MODE`); err != nil { + t.Fatal(err) + } + if _, err := first.Exec(ctx, `COMMIT`); err != nil { + t.Fatal(err) + } + + err := <-stopped + var pgErr *pgconn.PgError + if !errors.As(err, &pgErr) || pgErr.Code != "55P03" { + t.Fatalf("the run ended with %v, want the drop to hit the lock timeout", err) + } + + // The catalog half of the state first, because the record table is still + // held and reading it here would block this test behind its own holder. + if n := rowsIn(t, c, `SELECT count(*) FROM pg_class WHERE relname = $1`, name); n != 1 { + t.Fatalf("%s is gone, so the run did not stop where this test needs it to", name) + } + if n := rowsIn(t, c, `SELECT count(*) FROM pg_inherits i JOIN pg_class p ON p.oid = i.inhrelid + WHERE p.relname = $1`, name); n != 0 { + t.Fatalf("%s is still inherited, so the detach did not complete", name) + } + + if _, err := second.Exec(ctx, `ROLLBACK`); err != nil { + t.Fatal(err) + } + + if n := rowsIn(t, c, `SELECT count(*) FROM events_partition_drops + WHERE partition_name = $1 AND dropped_at IS NULL`, name); n != 1 { + t.Fatalf("no open record for %s, and that record is the only thing that knows it exists", name) + } + + dropped, err := Retain(ctx, dsn, time.Now(), 28*24*time.Hour, PartitionSpan{Behind: horizonBehind}) + if err != nil { + t.Fatalf("the run after the one that stopped: %v", err) + } + var finished *Dropped + for i := range dropped { + if dropped[i].Name == name { + finished = &dropped[i] + } + } + if finished == nil || finished.Rows != 5 { + t.Fatalf("the next run reported %v, want %s among them with its 5 rows", dropped, name) + } + if n := rowsIn(t, c, `SELECT count(*) FROM pg_class WHERE relname = $1`, name); n != 0 { + t.Fatalf("%s is still on disk after the run that was meant to finish it", name) + } + if n := rowsIn(t, c, `SELECT count(*) FROM events_partition_drops + WHERE partition_name = $1 AND dropped_at IS NOT NULL AND rows_dropped = 5`, name); n != 1 { + t.Fatalf("the record for %s was not closed with its count", name) + } +} + +// The other half of the same state, and the reachable one: the drop went through +// and the process ended before the stamp. The count is already in the record by +// then, because it is written before the drop, so what a killed run leaves is an +// open row with its number, not an open row with nothing. An earlier version of +// this test cleared the count as well and so described a state no run produces, +// which meant the line that handles the real one was never exercised. +func TestARecordLeftOpenAfterTheDropIsClosedWithItsCount(t *testing.T) { + dsn := withHistory(t) + c := conn(t, dsn) + ctx := context.Background() + old := olderWeek(6) + seed(t, c, old, 2) + name := partitionName(weekStart(old)) + + if _, err := Retain(ctx, dsn, time.Now(), 28*24*time.Hour, PartitionSpan{Behind: horizonBehind}); err != nil { + t.Fatalf("first run: %v", err) + } + // Exactly what a process killed between the drop and the stamp leaves: the + // count and the detach time are in, the stamp is not. + if _, err := c.Exec(ctx, `UPDATE events_partition_drops SET dropped_at = NULL + WHERE partition_name = $1`, name); err != nil { + t.Fatal(err) + } + + dropped, err := Retain(ctx, dsn, time.Now(), 28*24*time.Hour, PartitionSpan{Behind: horizonBehind}) + if err != nil { + t.Fatalf("the run that should close the record: %v", err) + } + if len(dropped) != 1 || dropped[0].Name != name { + t.Fatalf("the run reported %v, want just %s", dropped, name) + } + if dropped[0].Rows != 2 || !dropped[0].Counted { + t.Fatalf("the run reported %d rows counted=%v, want the 2 the record already held", + dropped[0].Rows, dropped[0].Counted) + } + var rows *int64 + var stamped *time.Time + if err := c.QueryRow(ctx, `SELECT rows_dropped, dropped_at FROM events_partition_drops + WHERE partition_name = $1`, name).Scan(&rows, &stamped); err != nil { + t.Fatal(err) + } + if stamped == nil { + t.Fatalf("%s is gone and its record is still open", name) + } + if rows == nil || *rows != 2 { + t.Fatalf("the record lost its count while being closed, it says %v", rows) + } +} + +// And the state no run produces, which is a partition dropped by hand beside an +// open record. There is nothing left to count, so the record is closed as it +// stands and the caller is told the number is not a measurement. Zero would be a +// measurement, and the whole point of the record is to keep "held nothing" and +// "nobody looked" apart. +func TestARecordWithNoCountIsClosedWithoutInventingOne(t *testing.T) { + dsn := withHistory(t) + c := conn(t, dsn) + ctx := context.Background() + old := olderWeek(6) + seed(t, c, old, 3) + name := partitionName(weekStart(old)) + + if _, err := Retain(ctx, dsn, time.Now(), 28*24*time.Hour, PartitionSpan{Behind: horizonBehind}); err != nil { + t.Fatalf("first run: %v", err) + } + if _, err := c.Exec(ctx, `UPDATE events_partition_drops + SET dropped_at = NULL, rows_dropped = NULL, detached_at = NULL + WHERE partition_name = $1`, name); err != nil { + t.Fatal(err) + } + + dropped, err := Retain(ctx, dsn, time.Now(), 28*24*time.Hour, PartitionSpan{Behind: horizonBehind}) + if err != nil { + t.Fatalf("the run that should close the record: %v", err) + } + if len(dropped) != 1 || dropped[0].Name != name { + t.Fatalf("the run reported %v, want just %s", dropped, name) + } + if dropped[0].Counted { + t.Fatal("the run reported a counted number for a partition nobody counted") + } + var rows *int64 + var stamped *time.Time + if err := c.QueryRow(ctx, `SELECT rows_dropped, dropped_at FROM events_partition_drops + WHERE partition_name = $1`, name).Scan(&rows, &stamped); err != nil { + t.Fatal(err) + } + if stamped == nil { + t.Fatalf("%s is gone and its record is still open", name) + } + if rows != nil { + t.Fatalf("the record says %d rows for a partition nobody counted, want null", *rows) + } +}