refactor(cubestore): Replace invalidate_tables_cache flag with generi…

[ovr]

…c post-commit callback API

Introduce a generic set_post_commit_callback on BatchPipe<S> that fires after a successful RocksDB commit on the RW-loop thread. This replaces the hard-coded invalidate_tables_cache boolean flag, which was a hack that also ran before the actual commit.

• BatchPipe is now generic over a store type S (defaults to ()), allowing callbacks to receive a typed store reference
• RocksMetaStore::write_operation passes &RocksMetaStore to callbacks, giving them direct access to cached_tables and other metastore state
• Moved cached_tables from RocksStore to RocksMetaStore where it belongs
• Cache invalidation now only happens when the commit actually succeeds
• RocksTable methods are generic over S so they work with any BatchPipe<S>

[claude]

Claude finished @ovr's task in 2m 19s —— View job

---

Code Review: Replace invalidate_tables_cache with post-commit callback API

• Read PR diff and changed files
• Review code quality, correctness, and design
• Check for security and performance concerns
• Check for testing and documentation
• Post review feedback

---

Overall Assessment

This is a well-motivated refactoring. The old invalidate_tables_cache flag was a code smell — a boolean on BatchPipe that hard-coupled it to metastore-specific cache logic, and worse, ran the invalidation before the RocksDB commit. The new design is cleaner and more correct.

What's Good

• Correctness fix: Cache invalidation now runs after a successful batch_write_rows(), not before. The old code could leave the cache invalidated even if the write failed — this is a genuine bug fix disguised as a refactor.
• Clean generics: BatchPipe<'a, S = ()> with a default type parameter is well done. Existing consumers (cache_rocksstore.rs) don't need to change their signatures, while RocksMetaStore gets typed callback access.
• Proper ownership move: Moving cached_tables from RocksStore to RocksMetaStore is the right call — it was metastore-specific state living in a generic store struct.
• Minimal blast radius: Only 4 files changed, and rocks_table.rs changes are purely mechanical (adding <S> generic parameter).

Concerns (see inline comments for details)

1. Silent callback drop (inline): If store is None but a callback was set, it's silently discarded. This is safe today but is a subtle trap for future callers. Suggest adding a warning log, or making the type system enforce it.

2. 11 identical closures (inline): The set_post_commit_callback(|metastore| { *metastore.cached_tables.lock().unwrap() = None; }) pattern is duplicated 11 times. Extracting it into a helper (which could reuse the existing reset_cached_tables() method) would reduce duplication and make future changes safer.

3. Callback panic safety (inline): The doc comment correctly says "Must not panic", but this is unenforceable. Consider wrapping the callback invocation in catch_unwind to protect the RW-loop thread.

4. No new tests: The post-commit callback mechanism itself has no unit test coverage. A test that verifies the callback fires after a successful write (and doesn't fire after a failed write) would validate the core correctness improvement this PR makes.

Minor Notes

• The #[derive(Clone)] on RocksMetaStore is safe today (all fields are Arc-wrapped), but widens the API surface. A comment noting it's intentional for the callback pathway would help future readers.
• The Ok(...) → bare expression change in update_location_download_size (removing the Ok(...) wrapper around update_with_res_fn) is a nice cleanup but unrelated to the callback refactoring — fine to include but worth noting it's a separate concern.

---

[claude]

Good doc comment. Two observations:

1. The "Must not panic" contract is important but unenforceable at compile time. Consider wrapping the callback invocation in std::panic::catch_unwind at the call site (in write_operation_impl) to protect the RW-loop thread from a misbehaving callback.

2. Since set_post_commit_callback silently overwrites a previously set callback, if a write operation closure contains branching logic where multiple paths could set a callback, only the last one wins. This is documented, but worth noting that it may be surprising if the API gains more diverse callers. An alternative would be to collect callbacks into a Vec and run them all, or return an error/log if a callback already exists.

[ovr]

@claude it looks incorrect, because RocksStoreRWLoop handles catch_unwind while execution function on RW loop.

[claude]

Claude finished @ovr's task in 25s —— View job

---

You're right, I apologize for the incorrect suggestion. Looking at RocksStoreRWLoop::new (rocks_store.rs:841):

match std::panic::catch_unwind(std::panic::AssertUnwindSafe(fun)) {
    Err(panic_payload) => {
        let restore_error = CubeError::from_panic_payload(panic_payload);
        log::error!("Panic during read write loop execution: {}", restore_error);
    }
    Ok(res) => { ... }
}

The entire RocksStoreRWLoopFn closure — which includes write_operation_impl and therefore the post-commit callback invocation — is already wrapped in catch_unwind. Adding another catch_unwind around the callback would be redundant. The "Must not panic" doc comment should instead just note that any panic will be caught and logged by the RW loop, but won't propagate to the caller.

Thanks for the correction.

[claude]

Minor observation: self.clone() here clones the entire RocksMetaStore on every write operation. Since all fields are Arc-wrapped this is just reference count bumps (cheap), so it's fine. But it's worth noting that adding #[derive(Clone)] to RocksMetaStore widens the API surface — if any non-Arc field is added in the future, this clone could become expensive. A comment or a targeted clone_for_callback() method might make the intent clearer.

Also, accessing self.store.rw_loop_default_cf directly (instead of going through a method) requires the field to be pub(crate) — that visibility change is visible in rocks_store.rs. This is fine but creates a tighter coupling between the two modules.

[ovr]

@claude how will it affect performance? As I can see, this struct containts 4 fields and all of them are under Arc(s), it should clone 4 Arc(s)?

[claude]

Claude finished @ovr's task in 34s —— View job

---

Performance impact of self.clone() on RocksMetaStore

• Read struct definition
• Analyze clone cost
• Post answer

---

Yes, you're right. RocksMetaStore has exactly 4 fields, all Arc-wrapped:

pub struct RocksMetaStore {
    store: Arc<RocksStore>,
    cached_tables: Arc<Mutex<Option<Arc<Vec<TablePath>>>>>,
    disk_space_cache: Arc<RwLock<Option<(HashMap<String, u64>, SystemTime)>>>,
    upload_loop: Arc<WorkerLoop>,
}

So self.clone() is 4 atomic reference count increments — no heap allocations, no deep copies. Each Arc::clone is a single fetch_add (or equivalent) on the refcount's AtomicUsize, typically ~1–3 ns on modern hardware.

This happens once per write_operation call, well before the actual RocksDB commit (batch_write_rows). RocksDB write + fsync will dominate by 3–6 orders of magnitude (~milliseconds vs. nanoseconds), so the clone cost is immeasurable in practice. No performance concern here.

The only theoretical concern with frequent Arc::clone across threads is cache-line contention on the refcount, but since this is on the single RW-loop thread, that doesn't apply either.

---

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 58.01%. Comparing base (34b1329) to head (20e692e).
⚠️ Report is 2 commits behind head on master.

Additional details and impacted files

@@            Coverage Diff             @@
##           master   #10683      +/-   ##
==========================================
+ Coverage   57.91%   58.01%   +0.10%     
==========================================
  Files         215      215              
  Lines       16646    16648       +2     
  Branches     3347     3347              
==========================================
+ Hits         9641     9659      +18     
+ Misses       6516     6499      -17     
- Partials      489      490       +1     

Flag	Coverage Δ
cube-backend	58.01% <ø> (+0.10%)	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[claude]

Claude finished @ovr's task in 2m 54s —— View job

---

Comprehensive Code Review: Post-commit callback API refactoring

• Read PR diff and all changed files
• Review code quality, correctness, and design patterns
• Check for security and performance concerns
• Review testing coverage and edge cases
• Check documentation and API surface
• Post detailed review feedback

---

Overall Assessment

Approve — This is a well-executed refactoring with a genuine correctness improvement at its core. The changes are clean, well-structured across 4 logical commits, and responsive to prior review feedback.

---

Code Quality & Correctness

The core correctness fix is solid. The old code ran invalidate_tables_cache before batch_write_rows() — meaning the cache could be invalidated even when the RocksDB write subsequently failed. The new design runs the callback after a successful batch_write_rows() (rocks_store.rs:1082-1086), which is the correct sequencing.

Generics are well-designed. BatchPipe<'a, S = ()> with a default type parameter is the right approach — it keeps RocksStore::write_operation backward-compatible (using S = ()) while allowing RocksMetaStore::write_operation to thread through S = RocksMetaStore. The meta_store_table_impl! macro continues to go through RocksStore::write_operation with S = (), which is correct since those macro-generated delete operations don't need cache invalidation.

Good response to review feedback. The follow-up commit (6300465) simplified store: Option<S> to store: S directly, which is cleaner — when S = (), the callback receives &() which is harmless, making the type system handle what was previously a runtime check.

The update_location_download_size cleanup (removing redundant Ok(...) wrapper around update_with_res_fn) is a correct simplification — update_with_res_fn already returns Result<IdRow<T>, CubeError>, matching the closure's return type.

The From<regex::Error> impl (lib.rs:217-221) enables ? on Regex::new() in mod.rs:4849 (previously .unwrap()). This is appropriate and consistent with the other From impls in lib.rs.

Remaining Suggestion: Extract Duplicated Callback

The 10 identical closures in production code remain:

batch_pipe.set_post_commit_callback(|metastore| {
    *metastore.cached_tables.lock().unwrap() = None;
});

This could be a one-liner helper that reuses reset_cached_tables():

impl BatchPipe<'_, RocksMetaStore> {
    pub fn invalidate_tables_cache_on_commit(&mut self) {
        self.set_post_commit_callback(|ms| ms.reset_cached_tables());
    }
}

This is a minor suggestion — the current form is perfectly correct and the duplication is low-risk since all 10 sites are doing exactly the same thing. But if more callback types are added in the future, having a pattern for named helpers would be nice.

Security

No security concerns. The callback mechanism operates entirely within the RW-loop thread. cached_tables is behind Arc<Mutex<...>> with standard locking. The pub(crate) visibility on rw_loop_default_cf is appropriately scoped.

Performance

• No regression. self.clone() at mod.rs:1529 is 4 Arc reference count increments (~4-12 ns total) — immeasurable against the RocksDB write/fsync cost.
• Callback runs in the hot path but is a single mutex lock + None assignment — effectively free.
• Improvement: Cache invalidation no longer happens on failed writes, avoiding unnecessary cache misses that could trigger redundant get_tables_with_path read operations.

Testing

Good test coverage added in commit ce91f66:

• test_post_commit_callback_on_success correctly validates both positive (callback fires on success) and negative (callback doesn't fire on error) paths.
• Uses AtomicBool with SeqCst ordering — correct for cross-thread verification.
• The existing schema_test and non_empty_schema_test were improved to use Result return types with ? instead of .unwrap() — better error messages on failure (commit 20e692e).

Minor gap: No test for the "callback overwrites previous callback" behavior, but this is documented in the doc comment and not critical.

Documentation

• The doc comment on set_post_commit_callback (rocks_store.rs:667-670) is clear and specifies the key constraints (overwrites previous, must not panic, must not block on async).
• As noted in the prior review discussion, the "Must not panic" is already enforced by the catch_unwind wrapper in RocksStoreRWLoop::new — so it's a correctness note rather than a safety requirement.

Commit Structure

Clean 4-commit progression:

1. 222d512 — Core refactoring (callback API, cached_tables move, generics)
2. ce91f66 — Unit test for callback mechanism
3. 6300465 — Simplification from Option<S> to S (review feedback)
4. 20e692e — Test quality improvement (unwrap → Result/?)

---