Global recovery purges dropped jobs' state tables before compute nodes finish reset; an un-abortable OverWindow actor then reads an empty table and panics (delta_btree_map lib.rs:231)

[yuhao-su]

Describe the bug

Global recovery unregisters the state tables of dropped/cancelled streaming jobs from the Hummock version (purge_state_table_from_hummock) before it resets the compute nodes. If an actor of a dropped job is still running at that moment and cannot be aborted promptly (its JoinHandle::abort() only lands at the next Pending, and a CPU-bound loop never yields), the actor keeps processing already-buffered chunks against a table that no longer exists in the version. The streaming read path does not report this: HummockReadVersion::update(CommittedSnapshot) silently swaps in the new version and PinnedVersion::levels(table_id) returns an empty iterator, so the read succeeds with zero rows.

For OverWindow this turns a legitimate Delete in the chunk into a "ghost delete" (the row it deletes is not in the freshly loaded, empty partition cache). DeltaBTreeMap's cursor only tolerates Change::Delete when the key also exists in the snapshot, so move_impl hits change.as_insert().unwrap() and the compute process aborts.

Sequence (global recovery, recovery_inner):

1. A DROP MATERIALIZED VIEW commits its catalog change, but its DropStreamingJobs barrier is stuck behind a heavy chunk in the job's OverWindow actor (whole-partition recompute, cache_policy = full).
2. A user runs RECOVER (or any other global recovery trigger). The pending drop command fails with adhoc recovery triggered.
3. reload_runtime_info_impl → apply_pre_applied_drop_cancel → purge_state_table_from_hummock → HummockManager::purge → unregister_table_ids. The version delta with removed_table_ids = {dropped job's tables} is committed and pushed to compute nodes via the hummock notification channel.
4. Only then does PartialGraphManager::recover → ControlStreamManager::recover → new_control_stream make each compute node run LocalBarrierWorker::reset() (abort_and_wait_actors + clear_shared_buffer). Meta waits for the InitResponse, so control stream reset elapsed=… equals the time it took to join the slowest actor.
5. The OverWindow actor is inside build_changes (while let Some(..) = cursor.next() loops with no .await), so the abort cannot land. It finishes the current partition, moves to the next partition in the same chunk, and extend_cache_to_boundary ("loading the whole partition into cache") reads the purged table → empty → panic.

Note that per-database recovery already does this in the right order: EnteringInitializing is only entered after the database reset responses have been collected, and reload_database_runtime_info (which unregisters via cleanup_dropped_streaming_jobs) runs after that. Only the global path purges first.

Observed on a production cluster running v3.0.2 (over_partition.rs:391 / :169 frames there; control stream reset elapsed=15.9s, await tree Epoch [!!! 280s] > Materialize > Project > OverWindow <== current) and reproduced locally on main (633d6d9). The ordering in recovery_inner is identical in v3.0.2 and main.

The same panic signature was reported before in #12632 and #14493; both were closed without this ordering being identified.

Error message/log

Local reproduction (single-node, debug build, main 633d6d9). mv's state tables are 10/11/12, its actor is 16.

23:41:31.904 ERROR meta    adhoc_recovery{error=adhoc recovery triggered}:recovery_attempt: risingwave_meta::stream::stream_manager: failed to run drop command error=adhoc recovery triggered
23:41:31.904 INFO  meta    adhoc_recovery{error=adhoc recovery triggered}: risingwave_meta::barrier::worker: recovery start!
23:41:31.933 DEBUG compute risingwave_storage::hummock::event_handler::hummock_event_handler: update to hummock version: 160
             (version delta 160: removed_table_ids = {10, 11, 12}, state_table_info_delta = {})
23:41:31.937 DEBUG compute risingwave_stream::task::barrier_worker::managed_state: force stopping actor 16
23:41:31.938 DEBUG compute risingwave_stream::task::barrier_worker::managed_state: join actor 16
             (actor 16 keeps running: thread samples show build_changes -> DeltaBTreeMap cursor + WindowStates::slide)
23:41:48.453 TRACE compute actor{actor_id=16}:...:executor{OverWindow 1000000003}: risingwave_stream::executor::over_window::over_partition: loading the whole partition into cache partition=OwnedRow([Some(Int32(1))])
             (this is the first read of partition p=1; table 11 is already purged, the read returns 0 rows)

thread 'rw-streaming' (52572522) panicked at src/utils/delta_btree_map/src/lib.rs:231:54:
called `Option::unwrap()` on a `None` value
   5: move_impl<...>                 at ./src/utils/delta_btree_map/src/lib.rs:231:54
   6: peek<...>                      at ./src/utils/delta_btree_map/src/lib.rs:169:9
   7: peek_next<...>                 at ./src/utils/delta_btree_map/src/lib.rs:154:14
   8: new<...>                       at ./src/utils/delta_btree_map/src/lib.rs:51:20
   9: find_affected_ranges           at ./src/stream/src/executor/over_window/over_partition.rs:510:35
  10: build_changes                  at ./src/stream/src/executor/over_window/over_partition.rs:174:58
  11: apply_chunk                    at ./src/stream/src/executor/over_window/general.rs:509:67

*** await tree context of current task ***
Actor 16: `CREATE MATERIALIZED VIEW mv AS SELECT p, o, v, SUM(v) OVER w AS s1, ... FROM t WINDOW w AS (PARTITION BY p ORDER BY o)`
  Epoch 11291753903423488 [!!! 18.036s]
    Materialize 1000000005 [770.152ms]
      Project 1000000004 [770.152ms]
        OverWindow 1000000003 [770.152ms]  <== current

The process then aborts (panic hook), i.e. the whole compute node goes down, not just the job being dropped.

To Reproduce

Single node, debug or release, e.g. risingwave single-node --parallelism 4. The heavy partition p = 0 has v = 0 everywhere so its recompute produces no output (keeps the actor CPU-bound without yielding on yield chunk); the light partition p = 1 is what gets read after the purge.

SET streaming_parallelism = 1;
CREATE TABLE t (p INT, o INT, v INT, PRIMARY KEY (o));
INSERT INTO t SELECT 0, i, 0 FROM generate_series(100001, 110000) g(i);   -- p0, 10K rows
INSERT INTO t SELECT 1, i, i FROM generate_series(1, 100) g(i);           -- p1, 100 rows
CREATE MATERIALIZED VIEW mv AS
  SELECT p, o, v,
         SUM(v) OVER w AS s1, SUM(v*2) OVER w AS s2, SUM(v*3) OVER w AS s3,
         MAX(v) OVER w AS s4, MIN(v) OVER w AS s5, AVG(v) OVER w AS s6
  FROM t WINDOW w AS (PARTITION BY p ORDER BY o);
FLUSH;
RECOVER;                                       -- start with empty partition caches
INSERT INTO t VALUES (0, 10021441, 0); FLUSH;  -- warm p0 into the cache; p1 stays uncached

Then, from a shell (the exact delay is not critical; 0.6 s worked reliably at 10K rows, where the p0 recompute takes ~15 s in a debug build):

psql -h 127.0.0.1 -p 4566 -U root -d dev -c "UPDATE t SET p = 0, v = 0 WHERE o = 50;"
#   one chunk: Insert at the front of p0 (heavy recompute, no output) + Delete from p1
psql -h 127.0.0.1 -p 4566 -U root -d dev -c "DROP MATERIALIZED VIEW mv;" &
#   catalog committed; DropStreamingJobs barrier is stuck behind the chunk
sleep 0.6
psql -h 127.0.0.1 -p 4566 -U root -d dev -c "RECOVER;"
#   drop command aborted; recovery purges mv's tables, then tries to reset the CN

The compute node panics as soon as the actor finishes the p0 recompute and loads p1 (~17 s later at 10K rows).

Control run: the same UPDATE without the DROP / RECOVER processes fine (p1 is loaded right after p0, the row moves between partitions, no panic). So the data and the OverWindow logic are consistent; the only difference is the purge landing before the actor is gone.

Useful logging for confirming the timeline:

RUST_LOG='info,risingwave_storage::hummock::event_handler=debug,risingwave_stream::executor::over_window=trace,risingwave_stream::task::barrier_worker=debug'

and risingwave ctl hummock list-version-deltas --start-version-delta-id <n> --num-epochs <m> to see the removed_table_ids delta.

Expected behavior

• A RECOVER (or any global recovery) that races with a pending DROP must not crash the compute node. The dropped job's tables should only be unregistered from Hummock once no actor can read or write them anymore, i.e. after the compute nodes have completed their reset.
• Independently, a streaming read on a table that has been removed from the committed version should surface as an error (as the batch path already does: table id {} has been dropped), not as an empty result. Silently reading "nothing" turns a lifecycle bug into an inconsistency panic deep inside an operator.
• The OverWindow / DeltaBTreeMap panic should carry context (partition key, row key) like the other consistency_panic! sites in over_window/general.rs, instead of a bare unwrap.

How did you deploy RisingWave?

Production: Kubernetes (RisingWave Cloud). Reproduction: risingwave single-node on macOS.

The version of RisingWave

• Production: v3.0.2
• Reproduction: main at 633d6d9 (3.2.0-alpha), debug build. Relevant code paths (recovery_inner, reload_runtime_info_impl, ControlStreamManager::recover, LocalBarrierWorker::reset, HummockReadVersion::update, over_partition.rs, delta_btree_map) were compared against v3.0.2 and are equivalent.

Additional context

Proposed fix, in three layers:

1. Root cause (meta). Move the purge out of reload_runtime_info_impl (src/meta/src/barrier/context/recovery.rs, the purge_state_table_from_hummock(...) call) and run it in recovery_inner (src/meta/src/barrier/worker.rs) after PartialGraphManager::recover(...) returns, i.e. after every connected worker has acknowledged reset(). Return the set of state table ids to keep as a field of BarrierWorkerRuntimeInfoSnapshot. Purge is GC, so it can be skipped (with a log) when unconnected_workers is non-empty and picked up by the next recovery. resolve_hummock_version_epochs only reads kept tables and can stay where it is. This makes the global path match the per-database path.
2. Defense in depth (storage). In HummockReadVersion::update's CommittedSnapshot arm (src/storage/src/hummock/store/version.rs), when state_table_info no longer contains self.table_id for a non-replicated instance, mark the read version as dropped and make read_filter_for_version return a HummockError instead of a valid empty read. The actor then fails with a StreamError and the normal failure report path handles it; no process abort. Normal drops are unaffected because actors exit on the DropStreamingJobs barrier before cleanup_dropped_streaming_jobs unregisters.
3. Diagnostics (stream). In ensure_delta_in_cache (over_partition.rs), the ghost-delete check is intentionally only applied in non-strict mode (feat(consistency): tolerate inconsistent stream in OverWindow #17168). In strict mode, run the same contains_key check and consistency_panic! with the partition key and row key instead of falling through to delta_btree_map's as_insert().unwrap(). Optionally have DeltaBTreeMap::new validate that every Delete key exists in the snapshot.

Optional hardening: build_changes has two while let Some(..) = cursor.next() loops with no await point, so abort() cannot interrupt a long recompute. A tokio::task::consume_budget().await every N rows (as dispatch.rs already does) would let the reset finish in milliseconds instead of seconds. On its own this does not close the race, so it is a complement to (1), not a replacement.