feat(data delete): reap ingestor bookkeeping rows with the table (RFC-0003 I6, backend#1209)

internal/cli/data.go

@@ -5,6 +5,7 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"strings"
 
 	"github.com/spf13/cobra"
 
@@ -165,7 +166,7 @@ func runDataIngest(ctx context.Context, out, errOut io.Writer, a runDataIngestAr
  // DROP/rm and then "succeed".
  plan := push.PlanTeardown(existingTable)
  rmSpin := a.Printer.Spinner(fmt.Sprintf("Removing the existing %q first", existingTable), "")
-_, terr := push.Teardown(ctx, cs, &push.SPDYExecutor{Config: resolved.RestConfig, Client: cs}, resolved.Namespace, plan, push.PodSpecOptions{
+tres, terr := push.Teardown(ctx, cs, &push.SPDYExecutor{Config: resolved.RestConfig, Client: cs}, resolved.Namespace, plan, push.PodSpecOptions{
  Namespace: resolved.Namespace,
  PVCClaimName: pvc.ClaimName,
  PVCMountPath: pvc.MountPath,
@@ -185,6 +186,13 @@ func runDataIngest(ctx context.Context, out, errOut io.Writer, a runDataIngestAr
  "first, then re-run this ingest. Nothing new was staged. (%w)",
  existingTable, existingTable, terr)}
  }
+ if !tres.BookkeepingCleaned {
+ // Same surfacing `data delete` does (Bugbot on the PR): the
+ // overwrite pre-clean runs the identical teardown, and a silent
+ // bookkeeping failure here would hide the same schema-drift
+ // regression on this path.
+ a.Printer.Warnf("Bookkeeping cleanup incomplete — the old table is gone, but its run-journal/salt rows may remain: %s", strings.Join(tres.BookkeepingErrs, "; "))
+ }
  a.Printer.Successf("Removed the old %q — ingesting the new data.", existingTable)
 	}
 

internal/cli/data_delete.go

@@ -212,7 +212,7 @@ undone — re-ingesting the data is the only way back.`)
  p.Newline()
  p.Successf("Dry-run — nothing was deleted.")
  if a.OutputJSON {
- writeDataDeleteJSON(a.JSONOut, "dry-run", resolved.Namespace, release.ReleaseName, plan, nil)
+ writeDataDeleteJSON(a.JSONOut, "dry-run", resolved.Namespace, release.ReleaseName, plan, nil, false)
  jsonEmitted = true
  }
  return nil
@@ -234,7 +234,7 @@ undone — re-ingesting the data is the only way back.`)
  // exit 0. One closure so the pair can't drift apart.
  declined := func() error {
  if a.OutputJSON {
- writeDataDeleteJSON(a.JSONOut, "declined", resolved.Namespace, release.ReleaseName, plan, nil)
+ writeDataDeleteJSON(a.JSONOut, "declined", resolved.Namespace, release.ReleaseName, plan, nil, false)
  jsonEmitted = true
  }
  return cleanCancel(p, "nothing was deleted.")
@@ -283,9 +283,15 @@ undone — re-ingesting the data is the only way back.`)
 
 	p.Newline()
 	p.Successf("Deleted %s.%s and %d PVC path(s).", plan.Database, plan.Table, len(res.RemovedPaths))
+	if !res.BookkeepingCleaned {
+ // Best-effort cleanup failed — say so, or a schema-drift regression
+ // (a renamed keying column) is indistinguishable from a legacy
+ // cluster without the bookkeeping tables (review, Saqlain).
+ p.Warnf("Bookkeeping cleanup incomplete — the table is gone, but its run-journal/salt rows may remain: %s", strings.Join(res.BookkeepingErrs, "; "))
+	}
 	p.Infof("The dataset's catalog metadata is kept as a record on tracebloc, marked unavailable — never removed.")
 	if a.OutputJSON {
- writeDataDeleteJSON(a.JSONOut, "deleted", resolved.Namespace, release.ReleaseName, plan, res.RemovedPaths)
+ writeDataDeleteJSON(a.JSONOut, "deleted", resolved.Namespace, release.ReleaseName, plan, res.RemovedPaths, res.BookkeepingCleaned)
  jsonEmitted = true
 	}
 	return nil
@@ -302,12 +308,17 @@ type dataDeleteJSON struct {
 	Table string `json:"table"` // the REAL (case-resolved) spelling, not the raw argument
 	PVCPaths []string `json:"pvc_paths"`
 	RemovedPaths []string `json:"removed_paths"`
+	// BookkeepingCleaned mirrors push.TeardownResult: whether the
+	// run-journal/salt rows were removed with the table. Always false for
+	// dry-run/declined — nothing was attempted, and a strict consumer must
+	// never read "cleanup happened" out of a run that deleted nothing.
+	BookkeepingCleaned bool `json:"bookkeeping_cleaned"`
 }
 
 // writeDataDeleteJSON serializes the delete result to w (stdout in
 // --output-json mode). Marshal errors are dropped: marshaling our own
 // struct can't fail in practice, and the exit code remains the contract.
-func writeDataDeleteJSON(w io.Writer, status, namespace, release string, plan push.TeardownPlan, removed []string) {
+func writeDataDeleteJSON(w io.Writer, status, namespace, release string, plan push.TeardownPlan, removed []string, bookkeepingCleaned bool) {
 	pvcPaths := plan.PVCPaths
 	if pvcPaths == nil {
  pvcPaths = []string{} // emit [] not null
@@ -316,13 +327,14 @@ func writeDataDeleteJSON(w io.Writer, status, namespace, release string, plan pu
  removed = []string{} // emit [] not null
 	}
 	res := dataDeleteJSON{
- Status: status,
- Namespace: namespace,
- Release: release,
- Database: plan.Database,
- Table: plan.Table,
- PVCPaths: pvcPaths,
- RemovedPaths: removed,
+ Status: status,
+ Namespace: namespace,
+ Release: release,
+ Database: plan.Database,
+ Table: plan.Table,
+ PVCPaths: pvcPaths,
+ RemovedPaths: removed,
+ BookkeepingCleaned: bookkeepingCleaned,
 	}
 	b, err := json.MarshalIndent(res, "", " ")
 	if err != nil {

internal/cli/testdata/golden/zz-all-strings.golden

@@ -60,6 +60,7 @@ screen. %s/%d are runtime placeholders.
 "%s — %s"
 "%s, … and %d more"
 "%s/%s"
+"%s: %v"
 "%s: %w"
 "%s=%s,%s=%s"
 "%v (policy: %v)"
@@ -96,6 +97,8 @@ screen. %s/%d are runtime placeholders.
 "Already signed out."
 "Applies to your next training run; a run already going keeps its size."
 "Ask one of these admins (or ask them to grant you access)"
+"Bookkeeping cleanup incomplete — the old table is gone, but its run-journal/salt rows may remain: %s"
+"Bookkeeping cleanup incomplete — the table is gone, but its run-journal/salt rows may remain: %s"
 "CPU cores for one run (1–%d)"
 "CSV %s has no columns"
 "Can't reach tracebloc from here."
@@ -140,6 +143,7 @@ screen. %s/%d are runtime placeholders.
 "Ctrl-C to cancel"
 "Ctrl-C to stop watching — the run keeps going on the cluster"
 "DB failures"
+"DELETE FROM `%s`.`%s` WHERE table_name='%s'"
 "DROP TABLE IF EXISTS `%s`.`%s`"
 "Datasets in %s (0)"
 "Datasets in %s — %d"
@@ -553,6 +557,7 @@ screen. %s/%d are runtime placeholders.
 "resource env"
 "restarted ≥%d times — check logs: %v"
 "root"
+"running mysql query: %w%s"
 "scanning the cluster for tracebloc clients: %w"
 "schema"
 "schema entry %q must be col:TYPE (e.g. age:INT,price:FLOAT)"

internal/push/list_detailed.go

@@ -288,7 +288,7 @@ func runMySQLQuery(ctx context.Context, exec Executor, namespace, pod, container
 	script := `mysql -uroot -p"$MYSQL_ROOT_PASSWORD" -N`
 	if err := exec.Exec(ctx, namespace, pod, container,
  []string{"sh", "-c", script}, strings.NewReader(query), &stdout, &stderr); err != nil {
- return "", fmt.Errorf("querying datasets: %w%s", err, stderrSuffix(&stderr))
+ return "", fmt.Errorf("running mysql query: %w%s", err, stderrSuffix(&stderr))
 	}
 	return stdout.String(), nil
 }

internal/push/teardown.go

@@ -66,6 +66,18 @@ func PlanTeardown(table string) TeardownPlan {
 type TeardownResult struct {
 	DroppedTable bool
 	RemovedPaths []string
+	// BookkeepingCleaned reports whether the ingestor's bookkeeping rows
+	// for the table (run-journal + pseudonymization salt) were deleted
+	// alongside it. Best-effort: false on clusters whose ingestor never
+	// created those tables — the teardown itself still succeeds.
+	BookkeepingCleaned bool
+	// BookkeepingErrs carries the per-table failure detail (which
+	// bookkeeping table, mysql's stderr folded into the error) so callers
+	// can SURFACE it: a silent false is indistinguishable from the
+	// schema-drift regression this cleanup exists to prevent — e.g. a
+	// renamed keying column would otherwise no-op invisibly, reopening
+	// the husk-row leak (review, Saqlain).
+	BookkeepingErrs []string
 }
 
 // Teardown performs the in-cluster teardown described by plan:
@@ -107,6 +119,39 @@ func Teardown(ctx context.Context, cs kubernetes.Interface, exec Executor, names
 	}
 	res.DroppedTable = true
 
+	// 1b. Best-effort bookkeeping cleanup (RFC-0003 I6 — tracebloc/backend#1209):
+	// the ingestor keeps one run-journal row per ingest and one
+	// pseudonymization-salt row per table; dropping the table alone
+	// strands them. Under per-ingestion tables (data-ingestors#408)
+	// every dataset is its own table, so every delete would leave one
+	// husk row of each kind — an unbounded slow leak. plan.Table passed
+	// ValidateTableName ([A-Za-z_][A-Za-z0-9_]*), so it cannot escape
+	// the single-quoted literal. Each DELETE runs separately and
+	// best-effort: either bookkeeping table may be absent on clusters
+	// that never ran a journal-aware ingestor, and these are metadata
+	// rows, not data — never fail a teardown whose DROP succeeded.
+	// The SQL rides runMySQLQuery — stdin, never a shell -e argument: the
+	// string literal's single quotes would terminate a single-quoted shell
+	// string and mysql would see an unquoted identifier, silently no-oping
+	// the DELETEs forever (Bugbot on the PR). Column contract, pinned
+	// against data-ingestors tracebloc_ingestor/database.py: BOTH
+	// bookkeeping tables key these rows by `table_name` —
+	// RUNS_TABLE tracebloc_ingest_runs (ingestor_id PK, table_name
+	// indexed via ix_tracebloc_ingest_runs_table)
+	// SALT_TABLE tracebloc_ingest_meta (table_name PK, salt)
+	// Each DELETE stays a separate best-effort call: batched on one stdin,
+	// a missing first table would abort the second (mysql stops on error).
+	res.BookkeepingCleaned = true
+	for _, bookkeeping := range []string{ingestRunsTable, ingestMetaTable} {
+ cleanupSQL := fmt.Sprintf("DELETE FROM `%s`.`%s` WHERE table_name='%s'",
+ plan.Database, bookkeeping, plan.Table)
+ if _, err := runMySQLQuery(ctx, exec, namespace, mysqlPod, mysqlContainer, cleanupSQL); err != nil {
+ res.BookkeepingCleaned = false
+ res.BookkeepingErrs = append(res.BookkeepingErrs,
+ fmt.Sprintf("%s: %v", bookkeeping, err))
+ }
+	}
+
 	// 2. rm the PVC dirs from an ephemeral stage-identity pod (see the
 	// doc note above + #259). The pod owns the staging files it
 	// deletes, so this works on hostPath and CSI.

internal/push/teardown_test.go

@@ -3,6 +3,8 @@ package push
 import (
 	"context"
 	"errors"
+	"fmt"
+	"io"
 	"strings"
 	"testing"
 
@@ -183,3 +185,103 @@ func TestCleanStaging_PodCreateFailureReturnsError(t *testing.T) {
  t.Errorf("rm ran (%v) despite the pod never being created", fe.gotCmd)
 	}
 }
+
+// TestTeardown_CleansBookkeepingRows pins the RFC-0003 I6 half of teardown
+// (tracebloc/backend#1209): after the DROP, the ingestor's run-journal and
+// salt rows for the table are deleted best-effort — and a failure there
+// never fails a teardown whose DROP already succeeded.
+func TestTeardown_CleansBookkeepingRows(t *testing.T) {
+	newCS := func() *fake.Clientset {
+ cs := fake.NewClientset(&corev1.Pod{
+ ObjectMeta: metav1.ObjectMeta{Name: "mysql-0", Namespace: "tracebloc"},
+ Spec: corev1.PodSpec{Containers: []corev1.Container{{Name: "mysql"}}},
+ Status: corev1.PodStatus{Phase: corev1.PodRunning},
+ })
+ readyOnNextGet(cs)
+ return cs
+	}
+	opts := PodSpecOptions{
+ Namespace: "tracebloc",
+ PVCClaimName: "client-pvc",
+ PVCMountPath: "/data/shared",
+ Table: "ds_0f2ab1de3c444e558f66778899aabbcc",
+	}
+	plan := PlanTeardown("ds_0f2ab1de3c444e558f66778899aabbcc")
+
+	t.Run("journal and salt rows are deleted for the dropped table", func(t *testing.T) {
+ rec := &recordingExecutor{}
+ res, err := Teardown(context.Background(), newCS(), rec, "tracebloc", plan, opts)
+ if err != nil {
+ t.Fatalf("Teardown: %v", err)
+ }
+ if !res.BookkeepingCleaned {
+ t.Error("BookkeepingCleaned = false, want true")
+ }
+ // The SQL must arrive on STDIN with its quoted literal intact —
+ // never through a shell -e argument, where the literal's single
+ // quotes would be eaten by the shell (Bugbot: the DELETEs would
+ // silently no-op forever).
+ var journal, salt bool
+ for _, call := range rec.calls {
+ if strings.Contains(strings.Join(call.cmd, " "), "DELETE FROM") {
+ t.Errorf("DELETE passed as a shell argument (%q) — must be fed on stdin", call.cmd)
+ }
+ stdin := string(call.stdin)
+ want := "WHERE table_name='" + plan.Table + "'"
+ if strings.Contains(stdin, "DELETE FROM") && strings.Contains(stdin, ingestRunsTable) && strings.Contains(stdin, want) {
+ journal = true
+ }
+ if strings.Contains(stdin, "DELETE FROM") && strings.Contains(stdin, ingestMetaTable) && strings.Contains(stdin, want) {
+ salt = true
+ }
+ }
+ if !journal {
+ t.Errorf("no stdin DELETE against %s with a quoted literal for %s observed", ingestRunsTable, plan.Table)
+ }
+ if !salt {
+ t.Errorf("no stdin DELETE against %s with a quoted literal for %s observed", ingestMetaTable, plan.Table)
+ }
+	})
+
+	t.Run("bookkeeping failure never fails the teardown", func(t *testing.T) {
+ rec := &recordingExecutor{failWhenStdinContains: "DELETE FROM"}
+ res, err := Teardown(context.Background(), newCS(), rec, "tracebloc", plan, opts)
+ if err != nil {
+ t.Fatalf("Teardown should tolerate bookkeeping failures, got: %v", err)
+ }
+ if !res.DroppedTable {
+ t.Error("DroppedTable = false, want true")
+ }
+ if res.BookkeepingCleaned {
+ t.Error("BookkeepingCleaned = true, want false when the DELETEs fail")
+ }
+ if len(res.RemovedPaths) == 0 {
+ t.Error("PVC rm did not run — bookkeeping failure must not short-circuit step 2")
+ }
+	})
+}
+
+// recordingExecutor records every Exec call (command AND stdin) and can
+// fail calls whose stdin matches a marker.
+type recordingExecutor struct {
+	calls []execCall
+	failWhenStdinContains string
+}
+
+type execCall struct {
+	pod, container string
+	cmd []string
+	stdin []byte
+}
+
+func (r *recordingExecutor) Exec(ctx context.Context, namespace, pod, container string, cmd []string, stdin io.Reader, stdout, stderr io.Writer) error {
+	var in []byte
+	if stdin != nil {
+ in, _ = io.ReadAll(stdin)
+	}
+	r.calls = append(r.calls, execCall{pod: pod, container: container, cmd: cmd, stdin: in})
+	if r.failWhenStdinContains != "" && strings.Contains(string(in), r.failWhenStdinContains) {
+ return fmt.Errorf("simulated bookkeeping failure")
+	}
+	return nil
+}