Avoid leaking sitrep rows during deletion (#10143)

Fixes #10131

In an attempt to avoid using transactions during the INSERT pathway
(see: #10133)
this PR instead tries to alter how DELETE works for sitreps.

Rather than using a two-step GC process (listing orphaned sitreps and
deleting
them - even while concurrent INSERT operations could be underway) this
PR
also enables the deletion of "rows within a sitrep, for which the sitrep
metadata
record no longer exists".

This allows us to delete rows that were previously leaked, and is also
now the
primary mechanism by which rows are removed: we delete the high-level
sitrep first, then remove all these newly-orphaned rows. This is still
performed
within a transaction on the DELETE pathway to avoid reading torn sitreps
during the loading phase (as originally appeared in
#9594).

Author: smklein

dev-tools/omdb/src/bin/omdb/nexus.rs

@@ -3459,8 +3459,10 @@ fn print_task_fm_sitrep_loader(details: &serde_json::Value) {
 
 fn print_task_fm_sitrep_gc(details: &serde_json::Value) {
     let SitrepGcStatus {
-        orphaned_sitreps_found,
         orphaned_sitreps_deleted,
+        sitrep_metadata_batches,
+        batch_size,
+        child_tables,
         errors,
     } = match serde_json::from_value::<SitrepGcStatus>(details.clone()) {
         Err(error) => {
@@ -3473,23 +3475,51 @@ fn print_task_fm_sitrep_gc(details: &serde_json::Value) {
         Ok(status) => status,
     };
 
-    pub const ORPHANS_FOUND: &str = "orphaned sitreps found:";
-    pub const ORPHANS_DELETED: &str = "orphaned sitreps deleted:";
-    pub const ERRORS: &str = "errors:";
-    pub const WIDTH: usize =
-        const_max_len(&[ERRORS, ORPHANS_FOUND, ORPHANS_DELETED]) + 1;
-    pub const NUM_WIDTH: usize = 4;
+    const BASE_WIDTH: usize = 40;
+    const NUM_WIDTH: usize = 4;
+
+    // Ensure columns stay aligned even if a child table name is long.
+    let width = child_tables
+        .keys()
+        // We're using two spaces here to match the
+        // "orphaned {name} rows deleted" format string.
+        // Removing "{name}" leaves a space on either side.
+        .map(|name| "orphaned  rows deleted:".len() + name.len() + 1)
+        .fold(BASE_WIDTH, |w, l| w.max(l));
+
     if !errors.is_empty() {
-        println!("{ERRICON}   {ERRORS:<WIDTH$}{:>NUM_WIDTH$}", errors.len());
+        println!(
+            "{ERRICON}   {:<width$}{:>NUM_WIDTH$}",
+            "errors:",
+            errors.len()
+        );
         for error in errors {
             println!("      > {error}")
         }
     }
 
-    println!("    {ORPHANS_FOUND:<WIDTH$}{orphaned_sitreps_found:>NUM_WIDTH$}");
+    println!("    {:<width$}{batch_size:>NUM_WIDTH$}", "batch size:");
+    println!(
+        "    {:<width$}{orphaned_sitreps_deleted:>NUM_WIDTH$}",
+        "orphaned sitreps deleted:"
+    );
     println!(
-        "    {ORPHANS_DELETED:<WIDTH$}{orphaned_sitreps_deleted:>NUM_WIDTH$}"
+        "    {:<width$}{sitrep_metadata_batches:>NUM_WIDTH$}",
+        "  batches:"
     );
+
+    if child_tables.is_empty() {
+        eprintln!("(!)   warning: no child tables reported (likely a bug)");
+    }
+
+    for (table_name, stats) in &child_tables {
+        println!(
+            "    {:<width$}{:>NUM_WIDTH$}",
+            format!("orphaned {table_name} rows deleted:"),
+            stats.rows_deleted,
+        );
+        println!("    {:<width$}{:>NUM_WIDTH$}", "  batches:", stats.batches,);
+    }
 }
 
 fn print_task_fm_rendezvous(details: &serde_json::Value) {

dev-tools/omdb/tests/successes.out

@@ -713,8 +713,15 @@ task: "fm_sitrep_gc"
   configured period: every <REDACTED_DURATION>s
   last completed activation: <REDACTED ITERATIONS>, triggered by <TRIGGERED_BY_REDACTED>
     started at <REDACTED_TIMESTAMP> (<REDACTED DURATION>s ago) and ran for <REDACTED DURATION>ms
-    orphaned sitreps found:      0
-    orphaned sitreps deleted:    0
+    batch size:                               1000
+    orphaned sitreps deleted:                    0
+      batches:                                   1
+    orphaned fm_alert_request rows deleted:      0
+      batches:                                   1
+    orphaned fm_case rows deleted:               0
+      batches:                                   1
+    orphaned fm_ereport_in_case rows deleted:    0
+      batches:                                   1
 
 task: "fm_sitrep_loader"
   configured period: every <REDACTED_DURATION>s
@@ -1346,8 +1353,15 @@ task: "fm_sitrep_gc"
   configured period: every <REDACTED_DURATION>s
   last completed activation: <REDACTED ITERATIONS>, triggered by <TRIGGERED_BY_REDACTED>
     started at <REDACTED_TIMESTAMP> (<REDACTED DURATION>s ago) and ran for <REDACTED DURATION>ms
-    orphaned sitreps found:      0
-    orphaned sitreps deleted:    0
+    batch size:                               1000
+    orphaned sitreps deleted:                    0
+      batches:                                   1
+    orphaned fm_alert_request rows deleted:      0
+      batches:                                   1
+    orphaned fm_case rows deleted:               0
+      batches:                                   1
+    orphaned fm_ereport_in_case rows deleted:    0
+      batches:                                   1
 
 task: "fm_sitrep_loader"
   configured period: every <REDACTED_DURATION>s