review: surface backpressure reject/force-flush stats, preserve restored bucket range

Addresses code-review findings on the backpressure change:

- StatsSnapshot/stats table now expose backpressure_rejected_total and
  backpressure_force_flush_total (were OTel-only). The change says "page
  on rejected", so operators must be able to see it via SELECT * FROM
  timefusion_stats when telemetry isn't wired — especially on first
  deploys.

- FlushableBucket carries the rows' real min/max timestamp; take captures
  it before resetting the bucket atomics and restore_taken_bucket replays
  it (monotonic widen). Previously restore collapsed the range to the
  bucket start, hiding restored rows from time-range pruning until the
  next insert. Regression test restore_taken_bucket_preserves_timestamp_range.

- DEFAULT_BUCKET_DURATION_MICROS 600s -> 300s to track the config default,
  so the test-only BUCKET_DURATION_MICROS const can't diverge from the
  process-global runtime value once any test pins the OnceLock.

- Clarify the intentional drop(commit_guard)-before-backoff in both the
  append and dedup retry paths so it isn't removed as apparently redundant
  (holding it across the sleep would serialize all writers behind one
  writer's backoff).

Author: tonyalaribe

src/buffered_write_layer.rs

@@ -110,32 +110,40 @@ fn quarantine_entry(quarantine_dir: &std::path::Path, entry: &WalEntry, kind: &s
 /// `snapshot_stats()` and rendered as rows by `timefusion.stats()`.
 #[derive(Debug, Clone)]
 pub struct StatsSnapshot {
-    pub mem_project_count:          usize,
-    pub mem_total_buckets:          usize,
-    pub mem_total_rows:             usize,
-    pub mem_total_batches:          usize,
-    pub mem_estimated_bytes:        usize,
-    pub reserved_bytes:             usize,
-    pub max_memory_bytes:           usize,
-    pub pressure_pct:               u32,
-    pub wal_files:                  usize,
-    pub wal_disk_bytes:             u64,
-    pub wal_shards_per_topic:       usize,
-    pub wal_known_topics:           usize,
-    pub bucket_duration_micros:     i64,
+    pub mem_project_count:              usize,
+    pub mem_total_buckets:              usize,
+    pub mem_total_rows:                 usize,
+    pub mem_total_batches:              usize,
+    pub mem_estimated_bytes:            usize,
+    pub reserved_bytes:                 usize,
+    pub max_memory_bytes:               usize,
+    pub pressure_pct:                   u32,
+    pub wal_files:                      usize,
+    pub wal_disk_bytes:                 u64,
+    pub wal_shards_per_topic:           usize,
+    pub wal_known_topics:               usize,
+    pub bucket_duration_micros:         i64,
     /// Age of the oldest bucket in MemBuffer (seconds, computed from
     /// `now - min(bucket.min_timestamp)`). None when MemBuffer is empty.
     /// Alerting target: alert at > 2× `flush_interval_secs`.
-    pub oldest_bucket_age_secs:     Option<u64>,
+    pub oldest_bucket_age_secs:         Option<u64>,
     /// Cumulative flush successes/failures since process start. Mirrors the
     /// OTel `timefusion.flush.completed`/`failed` counters so tests can
     /// assert without configuring OTel.
-    pub flush_completed_total:      u64,
-    pub flush_failed_total:         u64,
+    pub flush_completed_total:          u64,
+    pub flush_failed_total:             u64,
     /// Times an insert hit the memory hard limit and applied backpressure
     /// (synchronous flush-to-Delta) instead of rejecting. Sustained growth =
     /// ingest outpacing flush; the matching OTel counter is the alert target.
-    pub backpressure_engaged_total: u64,
+    pub backpressure_engaged_total:     u64,
+    /// Inserts rejected after the backpressure window expired without freeing
+    /// memory — Delta flush isn't keeping up. PAGE on any growth (data is still
+    /// in the WAL but ingest is now dropping). Mirrored from OTel so operators
+    /// can watch it via the stats table when telemetry isn't wired.
+    pub backpressure_rejected_total:    u64,
+    /// Open-bucket force-flush escalations (a single busy window was itself the
+    /// pressure). Sustained growth = windows too large for the budget.
+    pub backpressure_force_flush_total: u64,
 }
 
 #[derive(Debug, Default)]
@@ -193,6 +201,10 @@ struct CoalescedGroup {
     wal_positions:     Vec<Option<walrus_rust::WalPosition>>,
     /// Source bucket_ids; drained from MemBuffer after the combined commit succeeds.
     source_bucket_ids: Vec<i64>,
+    /// Min/max timestamp across absorbed buckets (Option so the derived Default's
+    /// 0 can't corrupt the min). Carried onto the combined FlushableBucket.
+    min_timestamp:     Option<i64>,
+    max_timestamp:     Option<i64>,
 }
 
 struct CombinedBucket {
@@ -225,6 +237,8 @@ impl CoalescedGroup {
         }
         // Merge per-shard positions (max).
         self.wal_positions = merge_wal_positions(std::mem::take(&mut self.wal_positions), b.wal_positions);
+        self.min_timestamp = Some(self.min_timestamp.map_or(b.min_timestamp, |m| m.min(b.min_timestamp)));
+        self.max_timestamp = Some(self.max_timestamp.map_or(b.max_timestamp, |m| m.max(b.max_timestamp)));
         self.source_bucket_ids.push(b.bucket_id);
     }
 
@@ -236,6 +250,8 @@ impl CoalescedGroup {
             wal_shard_counts,
             wal_positions,
             source_bucket_ids,
+            min_timestamp,
+            max_timestamp,
         } = self;
         // `absorb` is only called via `groups.entry(..).or_default().absorb(b)`
         // so `key` is always set by the time we collapse the group.
@@ -251,6 +267,8 @@ impl CoalescedGroup {
             row_count,
             wal_shard_counts,
             wal_positions,
+            min_timestamp: min_timestamp.unwrap_or(i64::MAX),
+            max_timestamp: max_timestamp.unwrap_or(i64::MIN),
         };
         CombinedBucket { combined, source_bucket_ids }
     }
@@ -267,38 +285,40 @@ pub type TantivyIndexCallback =
     Arc<dyn Fn(String, String, Vec<RecordBatch>, Vec<String>) -> futures::future::BoxFuture<'static, anyhow::Result<()>> + Send + Sync>;
 
 pub struct BufferedWriteLayer {
-    config:                     Arc<AppConfig>,
-    wal:                        Arc<WalManager>,
-    mem_buffer:                 Arc<MemBuffer>,
-    shutdown:                   CancellationToken,
-    delta_write_callback:       Option<DeltaWriteCallback>,
-    tantivy_index_callback:     Option<TantivyIndexCallback>,
-    background_tasks:           Mutex<Vec<JoinHandle<()>>>,
-    flush_lock:                 Mutex<()>,
-    reserved_bytes:             AtomicUsize, // Memory reserved for in-flight writes
-    pressure_notify:            Arc<Notify>, // Wakes flush task when pressure threshold crossed
+    config:                         Arc<AppConfig>,
+    wal:                            Arc<WalManager>,
+    mem_buffer:                     Arc<MemBuffer>,
+    shutdown:                       CancellationToken,
+    delta_write_callback:           Option<DeltaWriteCallback>,
+    tantivy_index_callback:         Option<TantivyIndexCallback>,
+    background_tasks:               Mutex<Vec<JoinHandle<()>>>,
+    flush_lock:                     Mutex<()>,
+    reserved_bytes:                 AtomicUsize, // Memory reserved for in-flight writes
+    pressure_notify:                Arc<Notify>, // Wakes flush task when pressure threshold crossed
     /// Notified at the end of every flush task iteration (success or failure).
     /// Test hook: lets E2E harnesses await actual completion of background work
     /// instead of racing wall-clock sleeps.
-    flush_tick_notify:          Arc<Notify>,
+    flush_tick_notify:              Arc<Notify>,
     /// Notified at the end of every eviction task iteration.
-    eviction_tick_notify:       Arc<Notify>,
+    eviction_tick_notify:           Arc<Notify>,
     /// Cumulative flush counters mirrored alongside OTel `record_flush`.
     /// OTel global metric state is opt-in (only initialized when telemetry is
     /// configured), so these atomics give the harness an in-process way to
     /// assert on what the global counters would be.
-    flush_completed_total:      AtomicU64,
-    flush_failed_total:         AtomicU64,
-    backpressure_engaged_total: AtomicU64,
+    flush_completed_total:          AtomicU64,
+    flush_failed_total:             AtomicU64,
+    backpressure_engaged_total:     AtomicU64,
+    backpressure_rejected_total:    AtomicU64,
+    backpressure_force_flush_total: AtomicU64,
     // Required for WAL replay of UPDATE/DELETE whose SQL references UDFs.
-    function_registry:          Arc<crate::functions::FnRegistry>,
+    function_registry:              Arc<crate::functions::FnRegistry>,
     /// Caps concurrent detached tantivy sidecar builds so a fast flush cycle
     /// (post-F4 — one build per (project, table) per cycle) can't fan out
     /// past S3 connection / memory limits when many tables flush together.
     /// FOLLOW-UP: handles aren't stored; graceful shutdown does not await
     /// in-flight tantivy uploads. Acceptable for now because the sidecar is
     /// best-effort and the index can be rebuilt from Delta on demand.
-    tantivy_spawn_sem:          Arc<tokio::sync::Semaphore>,
+    tantivy_spawn_sem:              Arc<tokio::sync::Semaphore>,
 }
 
 impl std::fmt::Debug for BufferedWriteLayer {
@@ -342,6 +362,8 @@ impl BufferedWriteLayer {
             flush_completed_total: AtomicU64::new(0),
             flush_failed_total: AtomicU64::new(0),
             backpressure_engaged_total: AtomicU64::new(0),
+            backpressure_rejected_total: AtomicU64::new(0),
+            backpressure_force_flush_total: AtomicU64::new(0),
             function_registry,
             // 16 is well above realistic per-cycle table fan-out for the
             // monoscope workload (~5 distinct table names) while still
@@ -507,6 +529,7 @@ impl BufferedWriteLayer {
                     last_mem = now_mem;
                     if std::time::Instant::now() >= deadline {
                         crate::metrics::record_backpressure_rejected();
+                        self.backpressure_rejected_total.fetch_add(1, Ordering::Relaxed);
                         error!(
                             "Write backpressure exhausted after {:?}: used={}MB still over hard limit — Delta flush is not freeing memory; rejecting (data remains in WAL)",
                             timeout,
@@ -563,6 +586,7 @@ impl BufferedWriteLayer {
             return Ok(());
         }
         crate::metrics::record_backpressure_force_flush();
+        self.backpressure_force_flush_total.fetch_add(1, Ordering::Relaxed);
         for (project_id, table_name, bucket_id) in self.mem_buffer.current_bucket_keys(current) {
             let Some(bucket) = self.mem_buffer.take_bucket_for_flush(&project_id, &table_name, bucket_id) else {
                 continue;
@@ -1428,6 +1452,8 @@ impl BufferedWriteLayer {
             flush_completed_total: self.flush_completed_total.load(Ordering::Relaxed),
             flush_failed_total: self.flush_failed_total.load(Ordering::Relaxed),
             backpressure_engaged_total: self.backpressure_engaged_total.load(Ordering::Relaxed),
+            backpressure_rejected_total: self.backpressure_rejected_total.load(Ordering::Relaxed),
+            backpressure_force_flush_total: self.backpressure_force_flush_total.load(Ordering::Relaxed),
         }
     }
 

src/database.rs

@@ -2715,8 +2715,11 @@ impl Database {
                         last_error = Some(e);
                         debug!("Delta write conflict detected, retrying... (attempt {}/{})", retry_count, max_retries);
 
-                        // Release the commit lock before backing off — peers
-                        // shouldn't queue behind this writer's sleep.
+                        // Intentionally release the commit lock BEFORE the
+                        // backoff sleep — do not remove. Holding it across the
+                        // sleep would serialize every other writer behind this
+                        // writer's backoff, converting commit contention into a
+                        // throughput collapse.
                         drop(commit_guard);
                         // Exponential backoff for better handling of concurrent writes
                         let backoff_ms = 100 * (2_u64.pow(retry_count.min(5)));
@@ -3328,6 +3331,9 @@ impl Database {
                         break;
                     }
                     Err(e) => {
+                        // Drop BEFORE the backoff sleep below — do not remove.
+                        // Holding the commit lock across the sleep would block
+                        // every concurrent append behind this dedup retry.
                         drop(commit_guard);
                         let msg = e.to_string();
                         // "Transaction failed" covers the conflict checker erroring

src/mem_buffer.rs

@@ -29,7 +29,12 @@ use crate::functions::FnRegistry;
 // longer = larger Delta files. Matches default flush interval for aligned boundaries.
 // Note: Timestamps before 1970 (negative microseconds) produce negative bucket IDs,
 // which is supported but may result in unexpected ordering if mixed with post-1970 data.
-const DEFAULT_BUCKET_DURATION_MICROS: i64 = 10 * 60 * 1_000_000;
+// Fallback when `set_bucket_duration_micros` is never called (i.e. unit tests
+// that build a MemBuffer directly). MUST track `d_bucket_duration_secs` in
+// config.rs — prod always overrides via bootstrap, but keeping the two in sync
+// avoids the test-only `BUCKET_DURATION_MICROS` const diverging from the
+// process-global runtime value once any test pins the OnceLock.
+const DEFAULT_BUCKET_DURATION_MICROS: i64 = 5 * 60 * 1_000_000;
 #[cfg(test)]
 const BUCKET_DURATION_MICROS: i64 = DEFAULT_BUCKET_DURATION_MICROS;
 
@@ -55,7 +60,7 @@ const MAX_BATCH_COUNT_PER_BUCKET: usize = 8;
 const MAX_BATCH_BYTES_FOR_COALESCE: usize = 4 * 1024 * 1024;
 
 /// Configured bucket window in microseconds. Set once at startup via
-/// `set_bucket_duration_micros`; defaults to 10 minutes when unset. Smaller
+/// `set_bucket_duration_micros`; defaults to 5 minutes when unset. Smaller
 /// windows free MemBuffer memory sooner (because the previous bucket becomes
 /// flushable sooner) at the cost of more, smaller Delta commits.
 pub fn bucket_duration_micros() -> i64 {
@@ -253,6 +258,12 @@ pub struct FlushableBucket {
     /// Written into Delta commit metadata so a crash between Delta commit
     /// and `advance_by_counts` can recover the cursor from Delta on restart.
     pub wal_positions:    Vec<Option<walrus_rust::WalPosition>>,
+    /// Actual min/max timestamp of the taken rows, captured before the source
+    /// bucket's atomics were reset. `restore_taken_bucket` replays these so a
+    /// restored bucket keeps its true time range (and stays visible to
+    /// time-range pruning) rather than collapsing to the bucket's start.
+    pub min_timestamp:    i64,
+    pub max_timestamp:    i64,
 }
 
 #[derive(Debug, Default)]
@@ -1161,6 +1172,8 @@ impl MemBuffer {
                     wal_positions,
                     row_count: bucket.row_count.load(Ordering::Relaxed),
                     wal_shard_counts,
+                    min_timestamp: bucket.min_timestamp.load(Ordering::Relaxed),
+                    max_timestamp: bucket.max_timestamp.load(Ordering::Relaxed),
                 });
             }
         }
@@ -1216,8 +1229,10 @@ impl MemBuffer {
         let wal_state = std::mem::take(&mut *wal_g);
         let freed = bucket.memory_bytes.swap(0, Ordering::Relaxed);
         let row_count = bucket.row_count.swap(0, Ordering::Relaxed);
-        bucket.min_timestamp.store(i64::MAX, Ordering::Relaxed);
-        bucket.max_timestamp.store(i64::MIN, Ordering::Relaxed);
+        // Capture the real range as we reset the sentinels so a restore (on
+        // Delta commit failure) can replay it instead of guessing bucket-start.
+        let min_timestamp = bucket.min_timestamp.swap(i64::MAX, Ordering::Relaxed);
+        let max_timestamp = bucket.max_timestamp.swap(i64::MIN, Ordering::Relaxed);
         drop(wal_g);
         drop(batches_g);
         drop(bucket_ref);
@@ -1249,6 +1264,8 @@ impl MemBuffer {
             row_count,
             wal_shard_counts: counts,
             wal_positions: positions,
+            min_timestamp,
+            max_timestamp,
         })
     }
 
@@ -1282,7 +1299,11 @@ impl MemBuffer {
         }
         bucket.memory_bytes.fetch_add(added, Ordering::Relaxed);
         bucket.row_count.fetch_add(b.row_count, Ordering::Relaxed);
-        bucket.update_timestamps(b.bucket_id * bucket_duration_micros());
+        // Replay the true range (monotonic widen) so restored rows stay visible
+        // to time-range pruning; concurrent inserts into the same open bucket
+        // are preserved since fetch_min/max only widens.
+        bucket.update_timestamps(b.min_timestamp);
+        bucket.update_timestamps(b.max_timestamp);
         drop(wal_g);
         drop(batches_g);
         self.estimated_bytes.fetch_add(added, Ordering::Relaxed);
@@ -2399,6 +2420,28 @@ mod tests {
         assert_eq!(total, 2, "no text_match preds → all rows returned");
     }
 
+    /// Regression: `restore_taken_bucket` (the Delta-commit-failure path of the
+    /// open-bucket force-flush) used to reset the bucket's min/max to the bucket
+    /// *start* (`bucket_id * duration`), hiding restored rows from time-range
+    /// pruning until the next insert. It must replay the rows' real range.
+    #[test]
+    fn restore_taken_bucket_preserves_timestamp_range() {
+        use crate::test_utils::test_helpers::{json_to_batch, test_span};
+        let buffer = MemBuffer::new();
+        let dur = bucket_duration_micros();
+        let ts = 7 * dur + 12_345; // mid-bucket — distinct from the bucket-start sentinel
+        let bucket_id = MemBuffer::compute_bucket_id(ts);
+        buffer.insert("p1", "otel_logs_and_spans", json_to_batch(vec![test_span("a", "svc", "p1")]).unwrap(), ts).unwrap();
+
+        let taken = buffer.take_bucket_for_flush("p1", "otel_logs_and_spans", bucket_id).expect("bucket taken");
+        assert_eq!((taken.min_timestamp, taken.max_timestamp), (ts, ts), "take must capture the real row range");
+
+        buffer.restore_taken_bucket(&taken); // simulate Delta commit failure
+        let again = buffer.take_bucket_for_flush("p1", "otel_logs_and_spans", bucket_id).expect("restored bucket present");
+        assert_eq!((again.min_timestamp, again.max_timestamp), (ts, ts), "restore must preserve the true range");
+        assert_ne!(again.min_timestamp, bucket_id * dur, "must not collapse to bucket start");
+    }
+
     #[test]
     fn test_bucket_partitioning() {
         let buffer = MemBuffer::new();

src/stats_table.rs

@@ -111,6 +111,16 @@ impl StatsTableProvider {
             ));
             rows.push(("buffered_layer", "pressure_pct".into(), s.pressure_pct.to_string()));
             rows.push(("buffered_layer", "backpressure_engaged_total".into(), s.backpressure_engaged_total.to_string()));
+            rows.push((
+                "buffered_layer",
+                "backpressure_rejected_total".into(),
+                s.backpressure_rejected_total.to_string(),
+            ));
+            rows.push((
+                "buffered_layer",
+                "backpressure_force_flush_total".into(),
+                s.backpressure_force_flush_total.to_string(),
+            ));
             rows.push(("wal", "files".into(), s.wal_files.to_string()));
             rows.push(("wal", "disk_bytes".into(), s.wal_disk_bytes.to_string()));
             rows.push(("wal", "disk_mb".into(), format!("{:.1}", s.wal_disk_bytes as f64 / (1024.0 * 1024.0))));