diff --git a/.config/nextest.toml b/.config/nextest.toml index 9ff3f9904..3e6480d82 100644 --- a/.config/nextest.toml +++ b/.config/nextest.toml @@ -124,9 +124,10 @@ threads-required = 'num-test-threads' # three tests, one of which the default filter already excludes. # # `pitr_restore` boots three servers in sequence and runs the restore three -# times, so it needs the same raised kill. +# times, so it needs the same raised kill. `crash_boot_slow_replay` holds each +# restart's WAL replay past the statement deadline on purpose. [[profile.default.overrides]] -filter = 'binary(crash_wal_truncation) | binary(crash_ilp_timeseries_write) | binary(pitr_restore)' +filter = 'binary(crash_wal_truncation) | binary(crash_ilp_timeseries_write) | binary(pitr_restore) | binary(crash_boot_slow_replay)' test-group = 'server-process-serial' threads-required = 'num-test-threads' slow-timeout = { period = "30s", terminate-after = 8 } @@ -150,7 +151,7 @@ slow-timeout = { period = "30s", terminate-after = 8 } # hide the cause rather than fix it. The tail this costs is bounded — roughly a # dozen crash/shutdown tests at about ten seconds each. [[profile.default.overrides]] -filter = 'binary(wal_direct_io) | binary(ilp_client_address) | binary(timeseries_write_events) | binary(crash_recovery) | binary(crash_recovery_overlays) | binary(crash_recovery_analytics) | binary(crash_resp_kv_write) | binary(crash_metadata_applier_wedge) | binary(crash_dropped_collection_reclaim) | binary(crash_purge_not_resurrected) | binary(crash_mid_replay) | binary(crash_checkpoint_corruption) | binary(crash_checkpoint_truncate_window) | binary(crash_refused_write_not_resurrected) | binary(crash_replay_fail_stop) | binary(crash_core_stall) | binary(crash_replay_stamp) | binary(crash_replay_stamp_calvin) | binary(calvin_hold_liveness) | binary(apply_pipeline_group_independence) | binary(crash_kv_atomic_autocommit) | test(/^cases::startup_failure::/) | test(/^cases::shutdown_in_flight::/) | test(/^cases::shutdown_budget::/) | test(/^cases::shutdown_abort_offender::/) | test(/^cases::shutdown_idempotent::/)' +filter = 'binary(wal_direct_io) | binary(ilp_client_address) | binary(timeseries_write_events) | binary(crash_recovery) | binary(crash_recovery_overlays) | binary(crash_recovery_analytics) | binary(crash_resp_kv_write) | binary(crash_metadata_applier_wedge) | binary(crash_dropped_collection_reclaim) | binary(crash_purge_not_resurrected) | binary(crash_mid_replay) | binary(crash_checkpoint_corruption) | binary(crash_checkpoint_truncate_window) | binary(crash_refused_write_not_resurrected) | binary(crash_replay_fail_stop) | binary(crash_core_stall) | binary(crash_replay_stamp) | binary(crash_replay_stamp_calvin) | binary(calvin_hold_liveness) | binary(apply_pipeline_group_independence) | binary(crash_kv_atomic_autocommit) | binary(crash_named_database_catalog_refs) | test(/^cases::startup_failure::/) | test(/^cases::shutdown_in_flight::/) | test(/^cases::shutdown_budget::/) | test(/^cases::shutdown_abort_offender::/) | test(/^cases::shutdown_idempotent::/)' test-group = 'server-process' threads-required = 'num-test-threads' diff --git a/Cargo.lock b/Cargo.lock index 3ffda025f..053dca615 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1465,27 +1465,27 @@ dependencies = [ [[package]] name = "cranelift-assembler-x64" -version = "0.136.4" +version = "0.136.5" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c9c587da2696fbaa58e7a2fa2e5bfdd365d0928219f36992304ac1f15e5bc254" +checksum = "249a61efca6a54425e027f5704a6f68a0232b4df527e05a3ef947d89dbdcb078" dependencies = [ "cranelift-assembler-x64-meta", ] [[package]] name = "cranelift-assembler-x64-meta" -version = "0.136.4" +version = "0.136.5" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f28a60027b7556b680b051651291053ec381edf4ffbc01784997ae51a64269bb" +checksum = "d5336a46e05db2d515440b1ce8a9363e7bc04b608c0f66cc98dddc5492a66fbf" dependencies = [ "cranelift-srcgen", ] [[package]] name = "cranelift-bforest" -version = "0.136.4" +version = "0.136.5" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "62bfdb846acebfd591d601994ea283c98d525ac0bc9a84ca3f2c36bcf5b69ff1" +checksum = "01b3675b13214270fddbd257d5fbf4109be8b02d92fc636fe96e2829514a7235" dependencies = [ "cranelift-entity", "wasmtime-internal-core", @@ -1493,9 +1493,9 @@ dependencies = [ [[package]] name = "cranelift-bitset" -version = "0.136.4" +version = "0.136.5" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "aa9f8cc1080739d5f72b4ee7e00aff7d589f1284beaf0a19472c07878be7f299" +checksum = "d7c1d67aa02b6e174bd2b165c113e94af9498fcc20e2892d0f4d5ab8c9618d09" dependencies = [ "serde", "serde_derive", @@ -1504,9 +1504,9 @@ dependencies = [ [[package]] name = "cranelift-codegen" -version = "0.136.4" +version = "0.136.5" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d85417d493ecec833b5134546b4a252c78cc0bd5be11729a35d25dcff9e4e70b" +checksum = "26fe6f3ec5344c32602c414e986dcbe699fa501a3e3e7b02f887d9d12b14ae42" dependencies = [ "bumpalo", "cranelift-assembler-x64", @@ -1535,9 +1535,9 @@ dependencies = [ [[package]] name = "cranelift-codegen-meta" -version = "0.136.4" +version = "0.136.5" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d01c8ed0098360bae69ad6b82952d4247516799ce8050d45ad168d6689f4eae6" +checksum = "96c5e6323d458486d95781defdb5016a1176fc911d8bc8e64b696adbda6e9044" dependencies = [ "cranelift-assembler-x64-meta", "cranelift-codegen-shared", @@ -1548,24 +1548,24 @@ dependencies = [ [[package]] name = "cranelift-codegen-shared" -version = "0.136.4" +version = "0.136.5" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "04f9b86d809cec9c4e88b201f1b3ed3b0347674c9ccfbb8c355228669e34be81" +checksum = "f1260aa93b0bd48e8e55c1c7fd9ff6a260f1a1bc46e1f0151a5055b124e5b15d" [[package]] name = "cranelift-control" -version = "0.136.4" +version = "0.136.5" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ce182038fb10b719841e65fdf5deeee258c684f63439f5be680c2be059274b58" +checksum = "50c179632a09850f81f1a1e9116cb09330c734a86689f3ee4c3f4ec81543acbd" dependencies = [ "arbitrary", ] [[package]] name = "cranelift-entity" -version = "0.136.4" +version = "0.136.5" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7bca812b03a7b942796391f25bc02aa70a1a007fee3ce2756490c788429e9c39" +checksum = "d8c071188cb1c3b1803e23ae415a8d5f58d741d3a4b26a79b83dd61921112eac" dependencies = [ "cranelift-bitset", "serde", @@ -1575,9 +1575,9 @@ dependencies = [ [[package]] name = "cranelift-frontend" -version = "0.136.4" +version = "0.136.5" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a0c30563d2aef95e8d1a00a920199c99117b350339599f0e9487a7a1cf5d59de" +checksum = "e2b85ba0a55d7685117b275fe6f36c819d438862e4223d28f0b97902460ff0ac" dependencies = [ "cranelift-codegen", "hashbrown 0.17.1", @@ -1588,15 +1588,15 @@ dependencies = [ [[package]] name = "cranelift-isle" -version = "0.136.4" +version = "0.136.5" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9c54184333a275698edb1d3bdff226131a25d47133d792aa716e47e32b8b7f05" +checksum = "20ed364edcaed769da7b2709d540f8f7d0ff257749ba193f475e22dba6a0178b" [[package]] name = "cranelift-native" -version = "0.136.4" +version = "0.136.5" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "eb67a4c8a3180b5f1d77dffcd6317c05386ced42e5a31c7004523a9d99dc85c1" +checksum = "418d98c3dce08f0d13526692bbd1888bff4c5f45207f5093d168b5a224fbc88c" dependencies = [ "cranelift-codegen", "libc", @@ -1605,9 +1605,9 @@ dependencies = [ [[package]] name = "cranelift-srcgen" -version = "0.136.4" +version = "0.136.5" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b533db1679b4b728568bd490da6533cf50974c48a4d7534e78f1e0e02bc39769" +checksum = "4f016b9994e39cad68107f33fa5b8dcee8fb7e38332bde3f97690e5e8abe1b10" [[package]] name = "crash-context" @@ -1715,9 +1715,9 @@ checksum = "460fbee9c2c2f33933d720630a6a0bac33ba7053db5344fac858d4b8952d77d5" [[package]] name = "crypto-bigint" -version = "0.7.8" +version = "0.7.10" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d726facc8b27e604391b232b970cdc19b5468c69f08b466b9734c759216c7415" +checksum = "9c035cb64cba03dad2fb921f3e218785e8cf9a41790da5eb3a1400f62fa2be11" dependencies = [ "cpubits", "ctutils", @@ -1783,9 +1783,9 @@ dependencies = [ [[package]] name = "ctutils" -version = "0.4.5" +version = "0.4.7" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "38d28fb6aa7dd680b3ffd1a3e64e115c177bdba63c0ae7f9cd48129ff05c4fa0" +checksum = "a2376ffd6f96d8a9db59875b4b6b0237d6900b5baf64279714e05493ce64828a" dependencies = [ "cmov", "subtle", @@ -2102,7 +2102,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" dependencies = [ "libc", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -3561,9 +3561,9 @@ dependencies = [ [[package]] name = "libz-sys" -version = "1.1.30" +version = "1.1.31" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f710a23e6dbf193214fd46ca56a9d6864e550abe86202184532ae7275e46de19" +checksum = "ff20308ecabd635a2dc1dd5834a86c462cca20efba1d5d3ddab1f8aea3455cb2" dependencies = [ "cc", "libc", @@ -5617,9 +5617,9 @@ dependencies = [ [[package]] name = "pulley-interpreter" -version = "49.0.4" +version = "49.0.5" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5cd3de43f4f4cff536de8902097edbdf749321b97dda548f36a37918eb85bba3" +checksum = "1d74df9e20cba0f6dd07f01c3140535a6b23e40aecccfcacfb921621b41d0acb" dependencies = [ "cranelift-bitset", "log", @@ -5629,9 +5629,9 @@ dependencies = [ [[package]] name = "pulley-macros" -version = "49.0.4" +version = "49.0.5" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8319a8fa5269b4895957f8795861c42822bd26218e495f70060a963323d94847" +checksum = "8415513bfa407c06de6f11dbe9437f2352e075a5520ff491a131166d3d90aed1" dependencies = [ "proc-macro2", "quote", @@ -5728,7 +5728,7 @@ dependencies = [ "once_cell", "socket2", "tracing", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -6296,7 +6296,7 @@ dependencies = [ "errno", "libc", "linux-raw-sys", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -6364,7 +6364,7 @@ dependencies = [ "security-framework", "security-framework-sys", "webpki-root-certs", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -7103,7 +7103,7 @@ dependencies = [ "getrandom 0.3.4", "once_cell", "rustix", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -7955,9 +7955,9 @@ dependencies = [ [[package]] name = "wasmtime" -version = "49.0.4" +version = "49.0.5" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "36abb25de5518707e5ba96b586366b09db168d2d4976cb9dac57265a317497ea" +checksum = "d45bf3b13781b9febb0eef42e1e145146c92ef891f345fd918e0e0fa1568c992" dependencies = [ "addr2line", "async-trait", @@ -7992,9 +7992,9 @@ dependencies = [ [[package]] name = "wasmtime-environ" -version = "49.0.4" +version = "49.0.5" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "92f16a4d09fb1fbf2dcfeddb9818e1517a258e13b505b3bfdadcf47a19bf31ca" +checksum = "833f00f3dc6128159284c5b71e211f24c97d45e052230a14f767fd830a6fb8b0" dependencies = [ "anyhow", "cpp_demangle", @@ -8023,15 +8023,15 @@ dependencies = [ [[package]] name = "wasmtime-internal-component-util" -version = "49.0.4" +version = "49.0.5" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4006009fc3598af4e895ab18c6c33d2c6b7c657b1bd0a755ca7ee1abef6f0c16" +checksum = "b168d39f2d6b896d89fe096afa4ccdd7a8e927b01469b590af84f02fca3126a1" [[package]] name = "wasmtime-internal-core" -version = "49.0.4" +version = "49.0.5" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "947a732ad659f42c10ec639706fc8cb1f068dd9a55eebde8a1f24dfd06dfa525" +checksum = "7ad86a65d12c6a9202584ceb658c9a20cf5b25c32eed61c9be5e5f863f88f0d9" dependencies = [ "hashbrown 0.17.1", "libm", @@ -8040,9 +8040,9 @@ dependencies = [ [[package]] name = "wasmtime-internal-cranelift" -version = "49.0.4" +version = "49.0.5" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "fbfb8b455d524939e03df42e3e749f8e26e26ba2acd8b08fa00b3bda55686750" +checksum = "a71f0fd1dd40e4e18d4a20cdba29ff8e0eb2c0f21b12c2eb5869e0fea789dd05" dependencies = [ "cranelift-codegen", "cranelift-control", @@ -8066,9 +8066,9 @@ dependencies = [ [[package]] name = "wasmtime-internal-fiber" -version = "49.0.4" +version = "49.0.5" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "bd8aa8711f68b2cdd94c2b85ba5c8317edaa1ac4ee0a0534b957123680239307" +checksum = "fa4dfa39e58735b24ce3d28bfc8c2b3041603cc8a941bb268395351beff90172" dependencies = [ "cc", "libc", @@ -8080,9 +8080,9 @@ dependencies = [ [[package]] name = "wasmtime-internal-jit-debug" -version = "49.0.4" +version = "49.0.5" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e873e4504faf50aa8ce255d6ea7210d87b39d5f379c2486782e2c5f179c70ca8" +checksum = "f2eb98d4486c7b0163447d77dea06f3cc87e5e5ba5fc93613f8ec42e27ea181d" dependencies = [ "cc", "wasmtime-internal-versioned-export-macros", @@ -8090,9 +8090,9 @@ dependencies = [ [[package]] name = "wasmtime-internal-jit-icache-coherence" -version = "49.0.4" +version = "49.0.5" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "197ab1fda5da958a5c0b8a97eb241beafc48ac568b82ef4f12f38d6866c74173" +checksum = "852cae234735d79db6c58fe0d04b32f4078ad01fdb3fd6d488ac366132ba9785" dependencies = [ "libc", "wasmtime-internal-core", @@ -8101,9 +8101,9 @@ dependencies = [ [[package]] name = "wasmtime-internal-unwinder" -version = "49.0.4" +version = "49.0.5" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5577f3a5a979a7c3b1b23c1e9f13e70a8583acc8828471ec56f282117fc4952b" +checksum = "b9289378c82b9db7e7cf933d6bb441ad968536ac48f3f38c55b7cf9d3e8d87e1" dependencies = [ "cranelift-codegen", "log", @@ -8113,9 +8113,9 @@ dependencies = [ [[package]] name = "wasmtime-internal-versioned-export-macros" -version = "49.0.4" +version = "49.0.5" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f6399c9c4050ed431cf66e21c05fadd6f9731db44476f452f8c7142b4597b883" +checksum = "cc5c60c732d99dbcc5d670b71ba6ccf2b015cd16fc828990eaad2c4b8805ebdb" dependencies = [ "proc-macro2", "quote", @@ -8195,7 +8195,7 @@ version = "0.1.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22" dependencies = [ - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -8599,27 +8599,27 @@ checksum = "29666d0abbfad1e3dc4dcf6144730dd3a3ab225bbbdac83319345b1b44ccfc1b" [[package]] name = "zstd" -version = "0.14.2" +version = "0.14.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "057cfd910cfac363a0ada849592624b4c9ff2e10bef504c3433810d78ed96f93" +checksum = "db3ae9bf90af6e40197bef767eeefc46ad6397420fe873830bd85e5fbbfb3c70" dependencies = [ "zstd-safe", ] [[package]] name = "zstd-safe" -version = "8.1.0" +version = "8.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "bdd44c6a7284e91f3717755b24315a302edd9153a01f753c3cba3d765e8eafac" +checksum = "ded520f16b29e1a9b771c722221b2b0a0398c2316d9bb3c42b458a5414d7434b" dependencies = [ "zstd-sys", ] [[package]] name = "zstd-sys" -version = "2.1.1+zstd.1.5.7" +version = "2.1.2+zstd.1.5.7" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "aeec9eaf2dffbbd09201e23bd0ffcbaa33bb8e9266a10734fd7ed90a85eca078" +checksum = "481ada9c543b2323cbb7bbdbb5249177fbf0188ed959ed2c15e4b22421c84ed3" dependencies = [ "cc", "pkg-config", diff --git a/nodedb-test-support/src/single_node.rs b/nodedb-test-support/src/single_node.rs index e2fcbe140..b952061bb 100644 --- a/nodedb-test-support/src/single_node.rs +++ b/nodedb-test-support/src/single_node.rs @@ -107,6 +107,8 @@ pub async fn start( let (sequencer, _gate) = StartupSequencer::new(); let gates = nodedb::bootstrap::cluster_ready::ClusterReadyGates { + data_plane_replay_gate: sequencer + .register_gate(StartupPhase::DataPlaneReplay, "data-plane-replay"), raft_gate: sequencer.register_gate(StartupPhase::RaftMetadataReplay, "raft"), schema_gate: sequencer.register_gate(StartupPhase::SchemaCacheWarmup, "schema"), sanity_gate: sequencer.register_gate(StartupPhase::CatalogSanityCheck, "sanity"), diff --git a/nodedb/src/bootstrap/cluster_ready.rs b/nodedb/src/bootstrap/cluster_ready.rs index 53544d387..fcfb93d03 100644 --- a/nodedb/src/bootstrap/cluster_ready.rs +++ b/nodedb/src/bootstrap/cluster_ready.rs @@ -16,6 +16,7 @@ use crate::control::state::SharedState; /// All readiness gates passed to [`await_cluster_ready`]. pub struct ClusterReadyGates { + pub data_plane_replay_gate: ReadyGate, pub raft_gate: ReadyGate, pub schema_gate: ReadyGate, pub sanity_gate: ReadyGate, @@ -37,6 +38,7 @@ pub async fn await_cluster_ready( data_group_recovery_timeout: DataGroupRecoveryTimeout, ) -> anyhow::Result<()> { let ClusterReadyGates { + data_plane_replay_gate, raft_gate, schema_gate, sanity_gate, @@ -46,6 +48,13 @@ pub async fn await_cluster_ready( health_loop_gate, gateway_enable_gate, } = gates; + // Every boot step below that dispatches to a Data Plane core waits for + // its WAL replay first: a core drains no request until its replay ends, + // and a request's deadline runs from its dispatch. Metadata post-apply + // hooks wait on the phase this fires (`DataPlaneReplay`). + await_data_plane_replay(data_plane_replay_done, &data_plane_replay_gate).await?; + data_plane_replay_gate.fire(); + // Boot-time readiness gate: wait until the metadata raft group has // applied its first entry on this node before opening any // client-facing listener. This eliminates the restart-window race @@ -113,46 +122,6 @@ pub async fn await_cluster_ready( } sanity_gate.fire(); - // Wait for every Data Plane core to finish `replay_all_wal` before opening - // the client gateway. Each core rebuilds its in-memory indexes (HNSW, etc.) - // from the WAL on its own thread; `/healthz` must not report ready until - // that is done, or a just-restarted node would serve queries against - // half-rebuilt indexes (e.g. an empty vector search). A dropped sender - // means a core panicked during open/replay, and a signalled error means - // its replay stopped short of the WAL — fail closed on both, exactly as - // the raft-readiness gate does, rather than open the gateway on a broken - // core. - const REPLAY_READY_TIMEOUT: Duration = Duration::from_secs(300); - let replay_wait = async { - for rx in data_plane_replay_done { - rx.await - .map_err(|_| { - anyhow::anyhow!( - "data plane core exited before signalling WAL replay completion" - ) - })? - .map_err(|e| anyhow::anyhow!("data plane core WAL replay failed: {e}"))?; - } - Ok::<_, anyhow::Error>(()) - }; - match tokio::time::timeout(REPLAY_READY_TIMEOUT, replay_wait).await { - Ok(Ok(())) => { - info!("all data plane cores completed WAL replay"); - } - Ok(Err(e)) => { - data_groups_gate.fail(format!("data plane WAL replay failed: {e}")); - return Err(e); - } - Err(_) => { - data_groups_gate.fail(format!( - "data plane WAL replay did not complete within {REPLAY_READY_TIMEOUT:?}" - )); - return Err(anyhow::anyhow!( - "data plane WAL replay timeout after {REPLAY_READY_TIMEOUT:?}" - )); - } - } - // WAL replay only rebuilds what the NodeDB WAL holds. Writes to // replicable collections are durable as entries in their data group's // Raft log instead, and their engine state comes back only when each @@ -313,6 +282,39 @@ pub async fn await_cluster_ready( Ok(()) } +/// Wait for every Data Plane core to finish its WAL replay, and fail `gate` +/// when one does not. +/// +/// Each core rebuilds its in-memory indexes (HNSW, etc.) from the WAL on its +/// own thread before it enters its event loop. A dropped sender means a core +/// panicked during open or replay, and a signalled error means its replay +/// stopped short of the WAL. Both fail closed. The wait has no time bound: a +/// replay ends in one of those signals, and its length grows with the WAL +/// tail, so a fixed bound refuses every boot of a node whose tail outgrew it. +async fn await_data_plane_replay( + replay_done: Vec>>, + gate: &ReadyGate, +) -> anyhow::Result<()> { + for rx in replay_done { + let replayed = match rx.await { + Ok(replayed) => replayed, + Err(_) => { + let reason = "data plane core exited before signalling WAL replay completion"; + gate.fail(reason); + return Err(anyhow::anyhow!(reason)); + } + }; + if let Err(error) = replayed { + gate.fail(format!("data plane WAL replay failed: {error}")); + return Err(anyhow::anyhow!( + "data plane core WAL replay failed: {error}" + )); + } + } + info!("all data plane cores completed WAL replay"); + Ok(()) +} + /// How often the stall check samples the applied index while waiting. const RAFT_READY_POLL_INTERVAL: Duration = Duration::from_secs(1); diff --git a/nodedb/src/control/array_sync/raft_apply/common.rs b/nodedb/src/control/array_sync/raft_apply/common.rs index 11b65bc21..10cef30ee 100644 --- a/nodedb/src/control/array_sync/raft_apply/common.rs +++ b/nodedb/src/control/array_sync/raft_apply/common.rs @@ -17,7 +17,7 @@ use crate::control::server::dispatch_utils::{ ChangeFeedOwner, SubmitWrite, WalDurability, WriteOrdering, submit_write, }; use crate::control::state::SharedState; -use crate::types::{DatabaseId, ReadConsistency, RequestId, TenantId, TraceId, VShardId}; +use crate::types::{DatabaseId, ReadConsistency, TenantId, TraceId, VShardId}; /// Identifies a committed Raft entry within the apply loop. /// @@ -258,21 +258,14 @@ pub(super) async fn ensure_array_open( let open_request_id = open_request.request_id; let mut open_rx = state.tracker.register(open_request_id); - let dispatch_result = match state.dispatcher.lock() { - Ok(mut d) => d.dispatch(open_request), - Err(poisoned) => poisoned.into_inner().dispatch(open_request), - }; - - // A dispatch refusal, such as a capacity limit, keeps its own class. - dispatch_result?; + // A committed entry waits out a full dispatcher queue rather than failing + // for local load. Any other dispatch refusal keeps its own class. + crate::control::server::dispatch_utils::dispatch_when_capacity_frees(state, open_request, None) + .await?; - await_data_plane( - async move { open_rx.recv().await.ok_or(()) }, - open_request_id, - "OpenArray", - ) - .await - .map(|_| ()) + await_data_plane(async move { open_rx.recv().await.ok_or(()) }, "OpenArray") + .await + .map(|_| ()) } /// Build a `Request` for an array apply/open with default deadline / priority. @@ -313,24 +306,22 @@ pub(super) fn build_array_request( } } -/// How long [`await_data_plane`] waits for the Data Plane's response. -const DATA_PLANE_AWAIT_TIMEOUT: Duration = Duration::from_secs(30); - -/// Await the Data Plane response to request `request_id`. An error status -/// becomes [`apply_refusal`]. A timeout is `crate::Error::DeadlineExceeded` -/// (`57014`). A closed channel is `crate::Error::Internal` with `op_label`. +/// Await the Data Plane response to a committed array write. An error status +/// becomes [`apply_refusal`]. A closed channel is `crate::Error::Internal` with +/// `op_label`. +/// +/// The wait has no bound. The core never expires a committed write, so a +/// timeout here would report a write the core still applies as not applied. pub(super) async fn await_data_plane( rx: impl std::future::Future>, - request_id: RequestId, op_label: &str, ) -> ProposeResult { - match tokio::time::timeout(DATA_PLANE_AWAIT_TIMEOUT, rx).await { - Ok(Ok(resp)) if resp.status == Status::Ok => Ok(AppliedWrite::from_response(&resp)), - Ok(Ok(resp)) => Err(apply_refusal(op_label, &resp)), - Ok(Err(_)) => Err(crate::Error::Internal { + match rx.await { + Ok(resp) if resp.status == Status::Ok => Ok(AppliedWrite::from_response(&resp)), + Ok(resp) => Err(apply_refusal(op_label, &resp)), + Err(()) => Err(crate::Error::Internal { detail: format!("{op_label}: response channel closed"), }), - Err(_) => Err(crate::Error::DeadlineExceeded { request_id }), } } @@ -351,7 +342,7 @@ pub(super) fn apply_refusal(op_label: &str, response: &Response) -> crate::Error mod tests { use super::*; use crate::bridge::envelope::{ErrorCode, Payload}; - use crate::types::Lsn; + use crate::types::{Lsn, RequestId}; fn refusal(code: Option) -> Response { Response { @@ -398,39 +389,26 @@ mod tests { detail: "not on this engine".into(), }; let response = refusal(Some(code.clone())); - let result = await_data_plane( - async move { Ok::<_, ()>(response) }, - RequestId::new(1), - "OpenArray", - ) - .await; + let result = await_data_plane(async move { Ok::<_, ()>(response) }, "OpenArray").await; match result { Err(crate::Error::DataPlane(kept)) => assert_eq!(kept, code), other => panic!("expected the typed refusal, got {other:?}"), } } - /// A local timeout is the typed deadline error, `57014` on pgwire. + /// A committed write's answer is awaited however late it arrives. #[tokio::test(start_paused = true)] - async fn a_local_timeout_is_a_typed_deadline() { + async fn a_late_answer_is_awaited() { + let mut ok = refusal(None); + ok.status = Status::Ok; let result = await_data_plane( - std::future::pending::>(), - RequestId::new(7), + async move { + tokio::time::sleep(Duration::from_secs(600)).await; + Ok::<_, ()>(ok) + }, "OpenArray", ) .await; - let error = match result { - Err(error) => error, - Ok(_) => panic!("a pending response must time out"), - }; - assert!( - matches!( - &error, - crate::Error::DeadlineExceeded { request_id } if *request_id == RequestId::new(7) - ), - "expected a typed deadline, got {error:?}" - ); - let (_, state, _) = crate::control::server::pgwire::types::error_to_sqlstate(&error); - assert_eq!(state, nodedb_types::error::sqlstate::QUERY_CANCELED.0); + assert!(result.is_ok(), "a late answer is the outcome: {result:?}"); } } diff --git a/nodedb/src/control/catalog_entry/post_apply/async_dispatch/dispatcher.rs b/nodedb/src/control/catalog_entry/post_apply/async_dispatch/dispatcher.rs index ecbdb489e..a652e905e 100644 --- a/nodedb/src/control/catalog_entry/post_apply/async_dispatch/dispatcher.rs +++ b/nodedb/src/control/catalog_entry/post_apply/async_dispatch/dispatcher.rs @@ -86,6 +86,13 @@ pub async fn run_post_apply_async_side_effects( entry: CatalogEntry, shared: Arc, ) -> crate::Result<()> { + // A core drains no request until its WAL replay ends, and each dispatch + // below runs its deadline from the moment it is sent. An entry applied + // during boot waits for the replay rather than timing out against it. + shared + .startup + .await_phase(crate::control::startup::StartupPhase::DataPlaneReplay) + .await?; match entry { CatalogEntry::PutCollection(stored) => { // AWAITED: Register completes before the applied-index watcher diff --git a/nodedb/src/control/cluster/metadata_applier/pending_ddl.rs b/nodedb/src/control/cluster/metadata_applier/pending_ddl.rs index f850f4fe8..8374efe87 100644 --- a/nodedb/src/control/cluster/metadata_applier/pending_ddl.rs +++ b/nodedb/src/control/cluster/metadata_applier/pending_ddl.rs @@ -150,6 +150,21 @@ impl MetadataCommitApplier { ); } } + // The reservation drained every descriptor it alters, and only the + // finalize's catalog entries would have ended those drains. A drain + // has no expiry, so the cancel ends them, on every replica. The + // preparation lease serialises DDL, so no other DDL drain exists on + // these descriptors. + let altered: Vec<_> = record + .objects + .iter() + .filter_map(|object| match object { + PendingDdlObject::Alter { entry, .. } => altered_descriptor(entry, &shared), + PendingDdlObject::Create { .. } => None, + }) + .map(|descriptor_id| (descriptor_id, nodedb_cluster::DrainOwner::Ddl)) + .collect(); + crate::control::lease::apply_drain_ends(&shared, &altered)?; let queued = queue_teardowns( catalog, teardown, @@ -250,6 +265,21 @@ fn object_entry(object: &PendingDdlObject) -> &MetadataEntry { } } +/// The descriptor a reserved alter drained, from its catalog entry. +fn altered_descriptor( + entry: &MetadataEntry, + shared: &crate::control::state::SharedState, +) -> Option { + let payload = match entry { + MetadataEntry::CatalogDdl { payload } + | MetadataEntry::CatalogDdlAudited { payload, .. } => payload, + _ => return None, + }; + let decoded = catalog_entry::decode(payload).ok()?; + crate::control::lease::descriptor_id_and_prior_version(&decoded, shared) + .map(|(descriptor_id, _)| descriptor_id) +} + /// `(database_id, tenant_id, name)` when `entry` is a collection create — /// the only shape that registers a Data Plane engine eagerly at DDL time. fn created_collection_target(entry: &MetadataEntry) -> Option { diff --git a/nodedb/src/control/cluster/recovery_check/integrity.rs b/nodedb/src/control/cluster/recovery_check/integrity.rs index 36f0fe836..afaec682b 100644 --- a/nodedb/src/control/cluster/recovery_check/integrity.rs +++ b/nodedb/src/control/cluster/recovery_check/integrity.rs @@ -123,13 +123,6 @@ pub fn verify_redb_integrity(catalog: &SystemCatalog) -> Vec { .iter() .map(|c| (c.database_id.as_u64(), c.tenant_id, c.name.clone())) .collect(); - // These legacy object families do not yet carry a database scope, so keep - // their existing tenant/name relationship checks separate from the - // database-scoped trigger check. - let legacy_collection_keys: HashSet<(u64, String)> = collections - .iter() - .map(|c| (c.tenant_id, c.name.clone())) - .collect(); let user_names: HashSet = users.iter().map(|u| u.username.clone()).collect(); let role_names: HashSet = roles.iter().map(|r| r.name.clone()).collect(); let owner_keys: HashSet<(String, u64, u64, String)> = owners @@ -230,9 +223,9 @@ pub fn verify_redb_integrity(catalog: &SystemCatalog) -> Vec { check_owner_users(&owners, &user_names, &mut violations); check_permission_grantees(&permissions, &user_names, &role_names, &mut violations); check_trigger_collections(&triggers, &collection_keys, &mut violations); - check_rls_collections(&rls, &legacy_collection_keys, &mut violations); + check_rls_collections(&rls, &collection_keys, &mut violations); check_materialized_view_sources(&materialized_views, &collection_keys, &mut violations); - check_change_stream_collections(&change_streams, &legacy_collection_keys, &mut violations); + check_change_stream_collections(&change_streams, &collection_keys, &mut violations); check_schedule_targets(&schedules, &collection_keys, &mut violations); violations @@ -240,9 +233,6 @@ pub fn verify_redb_integrity(catalog: &SystemCatalog) -> Vec { /// Database-scoped collection keys: `(database_id, tenant_id, name)`. type CollectionKeys = HashSet<(u64, u64, String)>; -/// Collection keys of the object families with no database scope: -/// `(tenant_id, name)`. -type LegacyCollectionKeys = HashSet<(u64, String)>; /// Owner row keys: `(object_type, database_id, tenant_id, name)`. type OwnerKeys = HashSet<(String, u64, u64, String)>; @@ -343,20 +333,25 @@ fn check_trigger_collections( } } -/// Check 5: every rls_policy.collection exists. +/// Check 5: every rls_policy.collection exists in the policy's database. +/// +/// A collection qualifier that does not match the policy's database is a +/// dangling reference too: it names no collection that database holds. fn check_rls_collections( rls: &[StoredRlsPolicy], - legacy_collection_keys: &LegacyCollectionKeys, + collection_keys: &CollectionKeys, violations: &mut Vec, ) { for p in rls { - let key = (p.tenant_id, p.collection.clone()); - if !legacy_collection_keys.contains(&key) { + let found = p.collection_key().is_ok_and(|key| { + collection_keys.contains(&(p.database_id, p.tenant_id, key.name().to_owned())) + }); + if !found { violations.push(Divergence::new(DivergenceKind::DanglingReference { from_kind: "rls_policy", - from_key: format!("{}:{}", p.tenant_id, p.name), + from_key: format!("{}:{}:{}", p.database_id, p.tenant_id, p.name), to_kind: "collection", - to_key: format!("{}:{}", p.tenant_id, p.collection), + to_key: format!("{}:{}:{}", p.database_id, p.tenant_id, p.collection), })); } } @@ -392,20 +387,21 @@ fn check_materialized_view_sources( /// tenant. fn check_change_stream_collections( change_streams: &[ChangeStreamDef], - legacy_collection_keys: &LegacyCollectionKeys, + collection_keys: &CollectionKeys, violations: &mut Vec, ) { for cs in change_streams { if cs.collection == "*" { continue; } - let key = (cs.tenant_id, cs.collection.clone()); - if !legacy_collection_keys.contains(&key) { + let database_id = cs.database_id.as_u64(); + let key = (database_id, cs.tenant_id, cs.collection.clone()); + if !collection_keys.contains(&key) { violations.push(Divergence::new(DivergenceKind::DanglingReference { from_kind: "change_stream", - from_key: format!("{}:{}", cs.tenant_id, cs.name), + from_key: format!("{database_id}:{}:{}", cs.tenant_id, cs.name), to_kind: "collection", - to_key: format!("{}:{}", cs.tenant_id, cs.collection), + to_key: format!("{database_id}:{}:{}", cs.tenant_id, cs.collection), })); } } @@ -507,6 +503,67 @@ mod tests { })); } + fn rls_policy(database_id: u64, collection: &str, display_collection: &str) -> StoredRlsPolicy { + StoredRlsPolicy { + tenant_id: 1, + database_id, + collection: collection.into(), + display_collection: display_collection.into(), + name: "own_rows".into(), + policy_type_tag: 0, + predicate_text: String::new(), + mode_tag: 0, + on_deny_json: "\"Silent\"".into(), + enabled: true, + created_by: "owner".into(), + created_at: 0, + } + } + + fn rls_violations(policy: &StoredRlsPolicy) -> Vec { + let directory = tempfile::tempdir().expect("create integrity catalog directory"); + let store = CredentialStore::open(&directory.path().join("system.redb")) + .expect("open integrity catalog"); + let catalog = store.catalog(); + let database_id = DatabaseId::new(1024); + let mut collection = StoredCollection::stamped_for_test(1, "pat", "owner"); + collection.database_id = database_id; + catalog + .put_collection(database_id, &collection) + .expect("store named database collection"); + catalog.put_rls_policy(policy).expect("store rls policy"); + verify_redb_integrity(catalog) + .into_iter() + .filter(|violation| { + matches!( + &violation.kind, + DivergenceKind::DanglingReference { + from_kind: "rls_policy", + .. + } + ) + }) + .collect() + } + + #[test] + fn rls_policy_on_a_named_database_collection_is_not_dangling() { + let violations = rls_violations(&rls_policy(1024, "1024/pat", "pat")); + assert!(violations.is_empty(), "{violations:?}"); + } + + #[test] + fn rls_policy_row_without_display_name_resolves_its_qualified_name() { + let violations = rls_violations(&rls_policy(1024, "1024/pat", "")); + assert!(violations.is_empty(), "{violations:?}"); + } + + #[test] + fn rls_policy_qualified_for_another_database_is_dangling() { + let violations = rls_violations(&rls_policy(1024, "2048/pat", "")); + assert_eq!(violations.len(), 1, "{violations:?}"); + } + #[test] fn builtin_role_detection() { assert!(is_builtin_role("superuser")); diff --git a/nodedb/src/control/lease/drain_hold.rs b/nodedb/src/control/lease/drain_hold.rs new file mode 100644 index 000000000..4ae8685c1 --- /dev/null +++ b/nodedb/src/control/lease/drain_hold.rs @@ -0,0 +1,151 @@ +// SPDX-License-Identifier: BUSL-1.1 + +//! Ownership of started descriptor drains until something else ends them. +//! +//! A drain has no expiry, so a holder that gives up must end it explicitly. +//! [`DrainHolds`] records each drain a caller started. The caller calls +//! [`DrainHolds::keep`] once the entry whose apply ends the drains is +//! committed, or a record that ends them on cancel exists. Every other exit +//! ends them: [`DrainHolds::end`] proposes each end and waits for it, and a +//! drop (an early return, a cancelled future) hands each end to the +//! background lease releaser. Ending a drain that is not active is a no-op on +//! every node, so ending one twice is harmless. + +use nodedb_cluster::{DescriptorId, DrainOwner}; + +use crate::control::state::SharedState; + +use super::releaser::ReleaseRequest; + +/// Drains a caller started and still owns. +#[must_use = "dropping the holds ends their drains"] +pub(crate) struct DrainHolds<'a> { + shared: &'a SharedState, + held: Vec<(DescriptorId, DrainOwner)>, +} + +impl<'a> DrainHolds<'a> { + pub(crate) fn new(shared: &'a SharedState) -> Self { + Self { + shared, + held: Vec::new(), + } + } + + /// Own `owner`'s drain on `descriptor_id`. Call it before proposing the + /// drain start: a start whose outcome is unknown can still apply. + pub(crate) fn hold(&mut self, descriptor_id: DescriptorId, owner: DrainOwner) { + self.held.push((descriptor_id, owner)); + } + + /// Stop owning the drains. Another path ends them. + pub(crate) fn keep(mut self) { + self.held.clear(); + } + + /// End every held drain and wait until each end applied here. An end + /// that does not apply goes to the background releaser, which retries it + /// and records a drain it never ends. + pub(crate) async fn end(mut self) { + for (descriptor_id, owner) in std::mem::take(&mut self.held) { + if let Err(error) = + super::end_drain_async(self.shared, descriptor_id.clone(), owner.clone()).await + { + tracing::debug!( + ?descriptor_id, + ?owner, + %error, + "descriptor drain end failed; the background releaser retries it" + ); + submit_end(self.shared, descriptor_id, owner); + } + } + } +} + +impl Drop for DrainHolds<'_> { + fn drop(&mut self) { + for (descriptor_id, owner) in std::mem::take(&mut self.held) { + submit_end(self.shared, descriptor_id, owner); + } + } +} + +fn submit_end(shared: &SharedState, descriptor_id: DescriptorId, owner: DrainOwner) { + shared + .lease_runtime + .releaser + .submit(ReleaseRequest::DrainEnd { + descriptor_id, + owner, + }); +} + +#[cfg(test)] +mod tests { + use std::time::Duration; + + use nodedb_cluster::DescriptorKind; + + use super::*; + + /// Holds dropped without `keep` or `end`, as an early return or a + /// cancelled future drops them, end their drains in the background. + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn dropped_holds_end_their_drains() { + let cluster = crate::control::cluster::test_one_node::boot().await; + let state = &cluster.state; + let id = DescriptorId::new(0, 7, DescriptorKind::Collection, "orders"); + super::super::drain_for_ddl_async( + state, + id.clone(), + 3, + crate::control::metadata_proposer::DEFAULT_DRAIN_TIMEOUT, + 0, + ) + .await + .expect("an unleased descriptor drains at once"); + assert!(state.lease_drain.is_draining(&id, 3)); + + let mut holds = DrainHolds::new(state); + holds.hold(id.clone(), DrainOwner::Ddl); + drop(holds); + + let deadline = tokio::time::Instant::now() + Duration::from_secs(5); + while state.lease_drain.is_draining(&id, 3) { + assert!( + tokio::time::Instant::now() < deadline, + "a dropped hold left its drain in force" + ); + tokio::time::sleep(Duration::from_millis(10)).await; + } + cluster.shutdown().await; + } + + /// Kept holds leave their drains to whatever ends them. + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn kept_holds_leave_their_drains() { + let cluster = crate::control::cluster::test_one_node::boot().await; + let state = &cluster.state; + let id = DescriptorId::new(0, 7, DescriptorKind::Collection, "kept"); + super::super::drain_for_ddl_async( + state, + id.clone(), + 3, + crate::control::metadata_proposer::DEFAULT_DRAIN_TIMEOUT, + 0, + ) + .await + .expect("an unleased descriptor drains at once"); + + let mut holds = DrainHolds::new(state); + holds.hold(id.clone(), DrainOwner::Ddl); + holds.keep(); + tokio::time::sleep(Duration::from_millis(200)).await; + assert!( + state.lease_drain.is_draining(&id, 3), + "a kept hold must not end its drain" + ); + cluster.shutdown().await; + } +} diff --git a/nodedb/src/control/lease/drain_propose_async.rs b/nodedb/src/control/lease/drain_propose_async.rs index 3d3d8334f..86a9eed8c 100644 --- a/nodedb/src/control/lease/drain_propose_async.rs +++ b/nodedb/src/control/lease/drain_propose_async.rs @@ -15,6 +15,7 @@ use crate::control::metadata_proposer::wait::await_metadata_apply; use crate::control::state::SharedState; use crate::error::Error; +use super::DrainHolds; use super::drain_propose::{POLL_INTERVAL, drain_start_entry, drained_or_timed_out, encode_drain}; /// Drain every lease on `id` at `version <= up_to_version` for a `Put*` DDL. @@ -22,8 +23,8 @@ use super::drain_propose::{POLL_INTERVAL, drain_start_entry, drained_or_timed_ou /// The drain is owned by [`DrainOwner::Ddl`], so the apply of the DDL's own /// catalog entry ends it and no other owner's drain. /// -/// Returns `Ok(())` once they have drained. Errors on timeout or propose -/// failure. +/// Returns `Ok(())` once they have drained, and the caller owns the drain +/// from then on. Errors on timeout or propose failure, with the drain ended. /// /// `own_holds` is how many of those refcount units the requesting transaction /// holds itself — `0` for a caller with no lease scope of its own. A @@ -75,12 +76,22 @@ async fn start_drain( if up_to_version == 0 { return Ok(()); } - propose_drain_async( + // `is_draining` has no expiry backstop. The hold owns the drain from + // before its start is proposed, so a start whose outcome is unknown, a + // timed-out wait, and a cancelled future all end it. Only a drained + // return hands the drain to the caller. + let mut hold = DrainHolds::new(shared); + hold.hold(id.clone(), owner.clone()); + if let Err(error) = propose_drain_async( shared, drain_start_entry(shared, &id, &owner, up_to_version, max_wait), "drain_start", ) - .await?; + .await + { + hold.end().await; + return Err(error); + } let deadline = Instant::now() + max_wait; let drained = loop { @@ -100,19 +111,16 @@ async fn start_drain( Err(error) => break Err(error), } }; - if let Err(error) = drained { - // `is_draining` has no expiry backstop, so this explicit propose is - // the only thing that clears the drain after a timeout. Its own - // errors are logged and dropped. - if let Err(cleanup_err) = end_drain_async(shared, id, owner).await { - tracing::warn!( - error = %cleanup_err, - "descriptor lease drain: cleanup propose failed after timeout" - ); + match drained { + Ok(()) => { + hold.keep(); + Ok(()) + } + Err(error) => { + hold.end().await; + Err(error) } - return Err(error); } - Ok(()) } /// End `owner`'s drain on `id` on every node, and wait until the end applies diff --git a/nodedb/src/control/lease/mod.rs b/nodedb/src/control/lease/mod.rs index 91ffcef99..f981b4877 100644 --- a/nodedb/src/control/lease/mod.rs +++ b/nodedb/src/control/lease/mod.rs @@ -24,6 +24,7 @@ pub mod admission; pub mod descriptor_lookup; pub mod drain; pub mod drain_apply; +mod drain_hold; pub mod drain_propose; pub mod drain_propose_async; pub mod gc; @@ -47,6 +48,7 @@ pub use descriptor_lookup::{ }; pub use drain::{DescriptorDrainTracker, DrainEntry}; pub use drain_apply::{apply_drain_ends, apply_drain_start}; +pub(crate) use drain_hold::DrainHolds; pub use drain_propose_async::{drain_for_ddl_async, drain_for_owner_async, end_drain_async}; pub use holders::{LeaseHolders, LeaseRevocation, revoke_if_fenced, revoke_on_release}; pub(super) use leader_wait::propose_and_wait; diff --git a/nodedb/src/control/lease/releaser.rs b/nodedb/src/control/lease/releaser.rs index 615d70e82..97544af2f 100644 --- a/nodedb/src/control/lease/releaser.rs +++ b/nodedb/src/control/lease/releaser.rs @@ -12,12 +12,14 @@ //! attempt failed, or the node shut down, costs latency only. A descriptor //! lease expires at its `expires_at`, and the renewal loop releases an idle //! one at expiry. A DDL preparation lease is reclaimed by the metadata leader -//! once its lease window passed. +//! once its lease window passed. A drain has no expiry: a drain end that is +//! never proposed leaves its descriptor drained until a later end or a +//! restart, so its failure is recorded where it is detected. use std::sync::{Arc, Mutex, Weak}; use std::time::Duration; -use nodedb_cluster::DescriptorId; +use nodedb_cluster::{DescriptorId, DrainOwner}; use tokio::sync::mpsc; use crate::control::shutdown::{ShutdownPhase, spawn_loop}; @@ -39,8 +41,14 @@ pub(crate) enum ReleaseRequest { /// Release this node's leases on these descriptors that no statement /// holds when the release runs. UnheldDescriptors(Vec), - /// Release the DDL preparation lease `token`. + /// Abandon the DDL preparation lease `token`: cancel its pending DDL + /// record, then release it. DdlPrepare { token: u64 }, + /// End `owner`'s drain on `descriptor_id`. + DrainEnd { + descriptor_id: DescriptorId, + owner: DrainOwner, + }, } /// The bounded queue of releases synchronous code hands off. @@ -146,11 +154,7 @@ async fn release_with_retry(shared: &Weak, request: &ReleaseRequest tracing::debug!(?request, attempt, %error, "lease release failed; retrying"); } Err(error) => { - tracing::warn!( - ?request, - %error, - "lease release failed on every attempt; the lease ends at its expiry" - ); + give_up(request, &error); return; } } @@ -160,6 +164,26 @@ async fn release_with_retry(shared: &Weak, request: &ReleaseRequest } } +/// Record a request every attempt failed to apply. A descriptor lease ends at +/// its expiry and the metadata leader reclaims a DDL preparation lease, but a +/// drain has no expiry backstop. +fn give_up(request: &ReleaseRequest, error: &Error) { + match request { + ReleaseRequest::UnheldDescriptors(_) => tracing::warn!( + ?request, + %error, + "lease release failed on every attempt; the lease ends at its expiry" + ), + ReleaseRequest::DdlPrepare { token } => { + crate::diag::ddl_cleanup_not_completed(error, "abandon_lease", *token); + } + ReleaseRequest::DrainEnd { + descriptor_id, + owner, + } => crate::diag::descriptor_drain_not_ended(error, descriptor_id, owner), + } +} + async fn release_once(shared: &SharedState, request: &ReleaseRequest) -> Result<(), Error> { match request { ReleaseRequest::UnheldDescriptors(ids) => { @@ -168,6 +192,10 @@ async fn release_once(shared: &SharedState, request: &ReleaseRequest) -> Result< ReleaseRequest::DdlPrepare { token } => { crate::control::metadata_proposer::release_ddl_prepare_token(shared, *token).await } + ReleaseRequest::DrainEnd { + descriptor_id, + owner, + } => super::end_drain_async(shared, descriptor_id.clone(), owner.clone()).await, } } diff --git a/nodedb/src/control/metadata_proposer/catalog.rs b/nodedb/src/control/metadata_proposer/catalog.rs index 8e6bbec37..c01fe8601 100644 --- a/nodedb/src/control/metadata_proposer/catalog.rs +++ b/nodedb/src/control/metadata_proposer/catalog.rs @@ -9,6 +9,7 @@ use nodedb_cluster::{METADATA_GROUP_ID, MetadataEntry, encode_entry}; use crate::control::catalog_entry::{self, CatalogEntry}; use crate::control::constraint_install::{ConstraintTarget, constraint_targets}; +use crate::control::lease::DrainHolds; use crate::control::propose_outcome::ProposeOutcome; use crate::control::state::SharedState; use crate::error::Error; @@ -85,9 +86,10 @@ async fn propose_and_apply_locally( let _local_ddl_guard = lock_ddl_preparation_async(shared).await; let lease = acquire_ddl_prepare_lease_async(shared, handle.as_ref()).await?; + let mut drains = DrainHolds::new(shared); let proposed = async { - let drained = drain_prior_version(shared, entry).await?; - let (stamped, targets) = stamp_or_end_drain(shared, entry, drained).await?; + drain_prior_version(shared, entry, &mut drains).await?; + let (stamped, targets) = stamp(shared, entry)?; let log_index = propose_prepared( shared, handle.as_ref(), @@ -99,13 +101,20 @@ async fn propose_and_apply_locally( Ok::<_, Error>((log_index, targets)) } .await; + // The applied entry ended the drain. Any failure ends it here: a drain has + // no wall-clock expiry. The end is proposed after the entry, so it is a + // no-op when an entry whose outcome was unknown applied after all. + match &proposed { + Ok(_) => drains.keep(), + Err(_) => drains.end().await, + } lease.release().await; let (log_index, targets) = proposed?; Ok((ProposeOutcome::Replicated { log_index }, targets)) } -/// Drain the prior version of the descriptor `entry` changes. Returns the -/// drained descriptor, whose drain the entry's apply ends. +/// Drain the prior version of the descriptor `entry` changes, and record the +/// drain in `drains`. The entry's apply ends it. /// /// Leases acquired at plan time are refcounted and held through execute; when /// the last in-flight query using a descriptor completes, its @@ -116,18 +125,20 @@ async fn propose_and_apply_locally( async fn drain_prior_version( shared: &SharedState, entry: &CatalogEntry, -) -> Result, Error> { + drains: &mut DrainHolds<'_>, +) -> Result<(), Error> { let Some((descriptor_id, prior_version)) = crate::control::lease::descriptor_id_and_prior_version(entry, shared) else { - return Ok(None); + return Ok(()); }; if prior_version == 0 { - return Ok(None); + return Ok(()); } + drains.hold(descriptor_id.clone(), nodedb_cluster::DrainOwner::Ddl); crate::control::lease::drain_for_ddl_async( shared, - descriptor_id.clone(), + descriptor_id, prior_version, DEFAULT_DRAIN_TIMEOUT, // No transactional lease scope of its own: this is a bare, @@ -135,13 +146,11 @@ async fn drain_prior_version( // alongside a buffered write to the same descriptor. 0, ) - .await?; - Ok(Some(descriptor_id)) + .await } /// Stamp `entry`, and name the collections whose constraint version the -/// stamped entry changes. A failure ends the drain of `drained` and returns -/// the error. +/// stamped entry changes. /// /// Freezes the descriptor_version / constraint_version / modification_hlc /// HERE, at propose time, so the value is computed exactly once from this @@ -149,45 +158,15 @@ async fn drain_prior_version( /// entry. Every node applies the frozen value without re-deriving it, which /// makes replay-from-log on restart and re-delivery during learner catch-up /// idempotent. -async fn stamp_or_end_drain( +fn stamp( shared: &SharedState, entry: &CatalogEntry, - drained: Option, ) -> Result<(CatalogEntry, Vec), Error> { let catalog = shared.credentials.catalog(); - let stamped = catalog_entry::descriptor_stamp::stamp(entry.clone(), &shared.hlc_clock, catalog) - .and_then(|stamped| { - let targets = constraint_targets(catalog, std::slice::from_ref(&stamped))?; - Ok((stamped, targets)) - }); - match stamped { - Ok(stamped) => Ok(stamped), - Err(error) => Err(end_unproposed_drain(shared, drained, error).await), - } -} - -/// End the drain a DDL started when it fails before its entry is proposed. -/// A drain has no wall-clock expiry, and no apply of this entry will end it. -/// Returns `error`, the failure that stopped the DDL. -async fn end_unproposed_drain( - shared: &SharedState, - drained: Option, - error: Error, -) -> Error { - if let Some(descriptor_id) = drained - && let Err(end) = crate::control::lease::end_drain_async( - shared, - descriptor_id, - nodedb_cluster::DrainOwner::Ddl, - ) - .await - { - tracing::warn!( - error = %end, - "metadata propose: the drain of a DDL that failed before its propose did not end" - ); - } - error + let stamped = + catalog_entry::descriptor_stamp::stamp(entry.clone(), &shared.hlc_clock, catalog)?; + let targets = constraint_targets(catalog, std::slice::from_ref(&stamped))?; + Ok((stamped, targets)) } /// Wrap `entry` as a `CatalogDdl` metadata entry. diff --git a/nodedb/src/control/metadata_proposer/catalog_batch.rs b/nodedb/src/control/metadata_proposer/catalog_batch.rs index 1e1407ceb..baaa331b4 100644 --- a/nodedb/src/control/metadata_proposer/catalog_batch.rs +++ b/nodedb/src/control/metadata_proposer/catalog_batch.rs @@ -6,6 +6,7 @@ use nodedb_cluster::{METADATA_GROUP_ID, MetadataEntry}; use crate::control::catalog_entry::{self, CatalogEntry}; use crate::control::constraint_install::{ConstraintTarget, constraint_targets}; +use crate::control::lease::DrainHolds; use crate::control::security::catalog::SystemCatalog; use crate::control::server::shared::session::ddl_buffer; use crate::control::state::SharedState; @@ -117,8 +118,10 @@ async fn propose_replicated( ) -> Result, Error> { let _local_ddl_guard = lock_ddl_preparation_async(shared).await; let lease = acquire_ddl_prepare_lease_async(shared, handle).await?; + let mut drains = DrainHolds::new(shared); let proposed: Result, Error> = async { - let Some((entries, targets)) = plan_drain_and_stamp(shared, plan).await? else { + let Some((entries, targets)) = plan_drain_and_stamp(shared, plan, &mut drains).await? + else { return Ok::<_, Error>(None); }; let mut wrapped = Vec::with_capacity(entries.len()); @@ -138,81 +141,51 @@ async fn propose_replicated( Ok::<_, Error>(Some((log_index, entries, targets))) } .await; + // The applied batch ended its drains. Any failure ends them here: a drain + // has no wall-clock expiry. The ends are proposed after the batch, so they + // are no-ops when a batch whose outcome was unknown applied after all. + match &proposed { + Ok(_) => drains.keep(), + Err(_) => drains.end().await, + } lease.release().await; proposed } /// Plan, drain, and stamp a batch, and name the collections whose constraint /// version the stamped batch changes. The caller holds the DDL preparation -/// lock and lease. `None` is a plan with no entries. -/// -/// The batch's apply ends every drain it started. A failure before the batch -/// is proposed ends them here: a drain has no wall-clock expiry. +/// lock and lease, and owns every drain recorded in `drains`. `None` is a plan +/// with no entries. async fn plan_drain_and_stamp( shared: &SharedState, plan: impl FnOnce(&SystemCatalog) -> crate::Result>, + drains: &mut DrainHolds<'_>, ) -> Result, Vec)>, Error> { let catalog = shared.credentials.catalog(); let entries = plan(catalog)?; if entries.is_empty() { return Ok(None); } - if let Err(error) = drain_batch(shared, &entries).await { - return Err(end_batch_drains(shared, &entries, error).await); - } + drain_batch(shared, &entries, drains).await?; let stamped = - catalog_entry::descriptor_stamp::stamp_batch(entries.clone(), &shared.hlc_clock, catalog) - .and_then(|stamped| { - let targets = constraint_targets(catalog, &stamped)?; - Ok((stamped, targets)) - }); - match stamped { - Ok(stamped) => Ok(Some(stamped)), - Err(error) => Err(end_batch_drains(shared, &entries, error).await), - } -} - -/// End the drain of every entry of a batch that failed before it was -/// proposed. Returns `error`, the failure that stopped the batch. -/// -/// Every node installed the drains, so each DDL drain ends through the -/// metadata group. -async fn end_batch_drains(shared: &SharedState, entries: &[CatalogEntry], error: Error) -> Error { - for entry in entries { - if let Err(end) = end_replicated_drain(shared, entry).await { - tracing::warn!( - kind = entry.kind(), - error = %end, - "metadata batch: the drain of an unapplied entry did not end" - ); - } - } - error + catalog_entry::descriptor_stamp::stamp_batch(entries, &shared.hlc_clock, catalog)?; + let targets = constraint_targets(catalog, &stamped)?; + Ok(Some((stamped, targets))) } -/// End, through the metadata group, the DDL drain [`drain_batch`] started for -/// `entry`. Ending a drain that is not active is a no-op on every node. -async fn end_replicated_drain(shared: &SharedState, entry: &CatalogEntry) -> Result<(), Error> { - match crate::control::lease::descriptor_id_and_prior_version(entry, shared) { - Some((descriptor_id, prior_version)) if prior_version > 0 => { - crate::control::lease::end_drain_async( - shared, - descriptor_id, - nodedb_cluster::DrainOwner::Ddl, - ) - .await - } - _ => Ok(()), - } -} - -/// Drain the prior version of every descriptor the batch changes. -async fn drain_batch(shared: &SharedState, entries: &[CatalogEntry]) -> Result<(), Error> { +/// Drain the prior version of every descriptor the batch changes, and record +/// each drain in `drains`. +async fn drain_batch( + shared: &SharedState, + entries: &[CatalogEntry], + drains: &mut DrainHolds<'_>, +) -> Result<(), Error> { for entry in entries { if let Some((descriptor_id, prior_version)) = crate::control::lease::descriptor_id_and_prior_version(entry, shared) && prior_version > 0 { + drains.hold(descriptor_id.clone(), nodedb_cluster::DrainOwner::Ddl); crate::control::lease::drain_for_ddl_async( shared, descriptor_id, diff --git a/nodedb/src/control/metadata_proposer/ddl_prepare.rs b/nodedb/src/control/metadata_proposer/ddl_prepare.rs index 172e74fa3..024550bcc 100644 --- a/nodedb/src/control/metadata_proposer/ddl_prepare.rs +++ b/nodedb/src/control/metadata_proposer/ddl_prepare.rs @@ -76,20 +76,34 @@ fn owner_step(shared: &SharedState, token: u64, deadline: Instant) -> Result Result<(), Error> { let handle = shared.metadata_raft_handle()?; - propose_metadata_and_wait_async( - shared, - handle.as_ref(), - &MetadataEntry::DdlPrepareRelease { token }, - ) - .await - .map(|_| ()) + abandon_token(shared, handle.as_ref(), token).await +} + +/// Cancel `token`'s pending DDL record, then release `token`'s lease. +/// +/// The cancel is proposed even when this node has no record for `token`: a +/// `DdlPendingPropose` whose outcome its proposer never learned can still +/// apply, and it applies while `token` owns the lease. Proposed after it, the +/// cancel applies after it, so the record never outlives the lease. Cancel and +/// release are both no-ops for a token with nothing to cancel or release. +pub(super) async fn abandon_token( + shared: &SharedState, + handle: &dyn MetadataRaftHandle, + token: u64, +) -> Result<(), Error> { + propose_metadata_and_wait_async(shared, handle, &MetadataEntry::DdlPendingCancel { token }) + .await?; + propose_metadata_and_wait_async(shared, handle, &MetadataEntry::DdlPrepareRelease { token }) + .await + .map(|_| ()) } /// The metadata-Raft-serialized descriptor preparation lease an async @@ -139,6 +153,23 @@ impl DdlPrepareLease<'_> { } } } + + /// Give the lease up after a failure: cancel any pending DDL record the + /// lease reserved, then release it, and wait until both applied here. + /// + /// For every exit that does not finalize. A failed abandon leaves the + /// lease unreleased, so its drop hands it to the background releaser, + /// which retries the same two steps. + pub(crate) async fn abandon(mut self) { + match abandon_token(self.shared, self.handle, self.token).await { + Ok(()) => self.released = true, + Err(error) => tracing::debug!( + token = self.token, + %error, + "metadata DDL lease abandon failed; the background releaser retries it" + ), + } + } } impl Drop for DdlPrepareLease<'_> { @@ -158,6 +189,18 @@ pub(crate) async fn acquire_ddl_prepare_lease_async<'a>( ) -> Result, Error> { let token = next_token(shared); let deadline = Instant::now() + DDL_PREPARE_WAIT; + // The guard owns `token` from before its first propose. An acquire whose + // outcome is unknown (a missed deadline, an unanswered leader forward, a + // cancelled future) can still apply and make `token` the owner, so every + // exit that does not return the lease drops this guard and releases the + // token. A release of a token that never owned the lease applies as a + // no-op, and it is proposed after every acquire of that token. + let lease = DdlPrepareLease { + shared, + handle, + token, + released: false, + }; loop { propose_metadata_and_wait_async( @@ -172,14 +215,7 @@ pub(crate) async fn acquire_ddl_prepare_lease_async<'a>( loop { match owner_step(shared, token, deadline)? { - OwnerStep::Acquired => { - return Ok(DdlPrepareLease { - shared, - handle, - token, - released: false, - }); - } + OwnerStep::Acquired => return Ok(lease), OwnerStep::Retry => break, OwnerStep::Wait => tokio::time::sleep(DDL_PREPARE_POLL).await, } @@ -241,4 +277,40 @@ mod tests { drop(state); cluster.shutdown().await; } + + /// An acquire cancelled at any point, before or after its propose + /// applied, leaves the lease free well before the leader's reclaim + /// window: the next acquire gets it at once. + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn a_cancelled_acquire_never_strands_the_lease() { + let cluster = crate::control::cluster::test_one_node::boot().await; + let state = Arc::clone(&cluster.state); + let handle = state.metadata_raft_handle().expect("metadata raft handle"); + for cancel_after_us in [0u64, 100, 500, 1_000, 2_000, 5_000, 10_000] { + let cancelled = tokio::time::timeout( + Duration::from_micros(cancel_after_us), + acquire_ddl_prepare_lease_async(&state, handle.as_ref()), + ) + .await; + match cancelled { + Ok(lease) => lease.expect("acquire").release().await, + Err(_elapsed) => {} + } + let lease = tokio::time::timeout( + Duration::from_secs(5), + acquire_ddl_prepare_lease_async(&state, handle.as_ref()), + ) + .await + .unwrap_or_else(|_| { + panic!( + "an acquire cancelled after {cancel_after_us}us left the lease held; the \ + next acquire waited for the reclaim" + ) + }) + .expect("acquire after a cancelled acquire"); + lease.release().await; + } + drop(state); + cluster.shutdown().await; + } } diff --git a/nodedb/src/control/metadata_proposer/ddl_reclaim.rs b/nodedb/src/control/metadata_proposer/ddl_reclaim.rs index dd9b07c0c..604324925 100644 --- a/nodedb/src/control/metadata_proposer/ddl_reclaim.rs +++ b/nodedb/src/control/metadata_proposer/ddl_reclaim.rs @@ -11,14 +11,12 @@ use std::sync::{Arc, Weak}; use std::time::Duration; -use nodedb_cluster::MetadataEntry; - use crate::control::shutdown::{ShutdownPhase, spawn_loop}; use crate::control::state::SharedState; use crate::error::Error; use super::ddl_owner::{current_owner, reclaim_cause}; -use super::ddl_prepare::propose_metadata_and_wait_async; +use super::ddl_prepare::abandon_token; use super::handle::MetadataRaftHandle; /// How often the leader checks the lease owner. @@ -86,22 +84,18 @@ async fn reclaim_if_due(shared: &Weak) -> bool { true } -/// Cancel `token`'s pending DDL record, when it has one, then release its -/// lease, and await both applies here. +/// Cancel `token`'s pending DDL record, then release its lease, and await +/// both applies here. /// -/// The cancel goes first, so a pending record never outlives its lease. Both -/// entries are idempotent, and the release applies only while `token` still -/// owns the lease, so a reclaim that races the owner's own release is a no-op. +/// The cancel goes first, so a pending record never outlives its lease. It is +/// proposed even when this node holds no record: the stuck owner's own +/// `DdlPendingPropose` can still be in flight. Both entries are idempotent, and +/// the release applies only while `token` still owns the lease, so a reclaim +/// that races the owner's own release is a no-op. pub(crate) async fn reclaim_ddl_prepare_lease( shared: &SharedState, handle: &dyn MetadataRaftHandle, token: u64, ) -> Result<(), Error> { - if shared.pending_ddl.contains(token) { - propose_metadata_and_wait_async(shared, handle, &MetadataEntry::DdlPendingCancel { token }) - .await?; - } - propose_metadata_and_wait_async(shared, handle, &MetadataEntry::DdlPrepareRelease { token }) - .await - .map(|_| ()) + abandon_token(shared, handle, token).await } diff --git a/nodedb/src/control/planner/catalog_adapter/adapter.rs b/nodedb/src/control/planner/catalog_adapter/adapter.rs index 79dee1d42..010aa120b 100644 --- a/nodedb/src/control/planner/catalog_adapter/adapter.rs +++ b/nodedb/src/control/planner/catalog_adapter/adapter.rs @@ -62,6 +62,10 @@ pub struct OriginCatalog { /// `currval` reads only from it, so one session never sees another /// session's allocation. `None` for planning with no session behind it. pub(super) session_sequences: Option>, + /// Plans for a Describe, never for execution: sequence functions check + /// their sequence exists and resolve to a typed placeholder, without + /// advancing a counter or recording a session value. + pub(super) describing: bool, } impl OriginCatalog { @@ -94,6 +98,7 @@ impl OriginCatalog { array_catalog, sequence_registry: None, session_sequences: None, + describing: false, } } @@ -120,6 +125,7 @@ impl OriginCatalog { array_catalog: shared.array_catalog.clone(), sequence_registry: Some(Arc::clone(&shared.sequence_registry)), session_sequences: None, + describing: false, } } @@ -134,6 +140,14 @@ impl OriginCatalog { self } + /// Plan for a Describe only. Sequence functions keep their result type and + /// their existence check, but advance nothing and record nothing, so + /// describing a statement never consumes a value its execution would. + pub fn describing(mut self) -> Self { + self.describing = true; + self + } + /// Bind the calling session's `currval` map to this adapter. pub fn with_session_sequences( mut self, diff --git a/nodedb/src/control/planner/catalog_adapter/sequence_access.rs b/nodedb/src/control/planner/catalog_adapter/sequence_access.rs index acea5ad60..b26482075 100644 --- a/nodedb/src/control/planner/catalog_adapter/sequence_access.rs +++ b/nodedb/src/control/planner/catalog_adapter/sequence_access.rs @@ -13,10 +13,17 @@ use crate::control::sequence::error_map::{map_sequence_error, undefined_sequence use super::adapter::OriginCatalog; +/// The value a sequence function takes while describing. Only its type +/// reaches the client. +const DESCRIBE_PLACEHOLDER: i64 = 0; + impl OriginCatalog { /// Advance `name` and record the value as this session's `currval`. pub(super) fn advance_sequence(&self, name: &str) -> Result { let registry = self.require_sequence_registry()?; + if self.describing { + return self.describe_placeholder(registry, name); + } let database_id = self.database_id.as_u64(); let value = registry .nextval(database_id, self.tenant_id, name) @@ -32,6 +39,9 @@ impl OriginCatalog { // The sequence must exist before its absence from the session map can // mean "not yet called in this session" rather than "no such object". let registry = self.require_sequence_registry()?; + if self.describing { + return self.describe_placeholder(registry, name); + } if !registry.exists(self.database_id.as_u64(), self.tenant_id, name) { return Err(undefined_sequence(name)); } @@ -49,11 +59,28 @@ impl OriginCatalog { /// Position `name` so the next `nextval` returns `value + increment`. pub(super) fn position_sequence(&self, name: &str, value: i64) -> Result { let registry = self.require_sequence_registry()?; + if self.describing { + return self.describe_placeholder(registry, name).map(|_| value); + } registry .setval(self.database_id.as_u64(), self.tenant_id, name, value) .map_err(|e| map_sequence_error(name, e)) } + /// What a sequence function resolves to while describing: the same type + /// as its real value, for a sequence that exists. + fn describe_placeholder( + &self, + registry: &SequenceRegistry, + name: &str, + ) -> Result { + if registry.exists(self.database_id.as_u64(), self.tenant_id, name) { + Ok(DESCRIBE_PLACEHOLDER) + } else { + Err(undefined_sequence(name)) + } + } + fn require_sequence_registry(&self) -> Result<&SequenceRegistry, SqlError> { self.sequence_registry .as_deref() diff --git a/nodedb/src/control/planner/context/catalog_inputs.rs b/nodedb/src/control/planner/context/catalog_inputs.rs index 058c66a62..c1e4deb61 100644 --- a/nodedb/src/control/planner/context/catalog_inputs.rs +++ b/nodedb/src/control/planner/context/catalog_inputs.rs @@ -26,6 +26,16 @@ pub(super) struct CatalogInputs { } impl CatalogInputs { + /// The node-wide sequence counters these inputs plan against, if any. + pub(super) fn sequence_registry( + &self, + ) -> Option> { + match self.shared.as_ref().and_then(|weak| weak.upgrade()) { + Some(shared) => Some(Arc::clone(&shared.sequence_registry)), + None => self.sequence_registry.clone(), + } + } + pub(super) fn build_adapter( &self, tenant_id: u64, diff --git a/nodedb/src/control/planner/context/query/planning.rs b/nodedb/src/control/planner/context/query/planning.rs index ef1ea582d..f32da9385 100644 --- a/nodedb/src/control/planner/context/query/planning.rs +++ b/nodedb/src/control/planner/context/query/planning.rs @@ -99,14 +99,20 @@ impl QueryContext { let catalog = if purpose == PlanningPurpose::Metadata { // Metadata requests intentionally do not participate in descriptor // lease admission; they only need a stable catalog snapshot for - // authorization and response shaping. - crate::control::planner::catalog_adapter::OriginCatalog::new( + // authorization and response shaping. They never execute, so + // sequence functions resolve to their type and advance nothing. + let catalog = crate::control::planner::catalog_adapter::OriginCatalog::new( Arc::clone(&inputs.credentials), inputs.array_catalog.clone(), tenant_id.as_u64(), database_id, inputs.retention_policy_registry.clone(), ) + .describing(); + match inputs.sequence_registry() { + Some(registry) => catalog.with_sequence_registry(registry), + None => catalog, + } } else { inputs.build_adapter(tenant_id.as_u64(), database_id) }; diff --git a/nodedb/src/control/security/catalog/rls.rs b/nodedb/src/control/security/catalog/rls.rs index b252fc82d..e1f27d17c 100644 --- a/nodedb/src/control/security/catalog/rls.rs +++ b/nodedb/src/control/security/catalog/rls.rs @@ -18,6 +18,7 @@ //! policy whatever happens: a row that cannot be compiled installs as a //! restrictive deny-all so it can never widen access. +use nodedb_types::{CollectionKey, CollectionKeyError}; use redb::{ReadableDatabase, ReadableTable, TableDefinition}; use tracing::error; @@ -200,6 +201,22 @@ impl StoredRlsPolicy { } } + /// The catalog key of the policy's collection: its database and bare name. + /// + /// A row written before `display_collection` existed carries only the + /// qualified `collection`, so the qualifier is stripped from that. + pub(crate) fn collection_key(&self) -> Result, CollectionKeyError> { + let database_id = DatabaseId::new(self.database_id); + if self.display_collection.is_empty() { + CollectionKey::from_qualified_str(database_id, &self.collection) + } else { + Ok(CollectionKey::from_bare( + database_id, + &self.display_collection, + )) + } + } + /// The unqualified collection name, falling back to `collection`. fn display_name(&self) -> &str { if self.display_collection.is_empty() { diff --git a/nodedb/src/control/server/dispatch_utils/minted/owned.rs b/nodedb/src/control/server/dispatch_utils/minted/owned.rs index 3e045811d..7dcb1efd5 100644 --- a/nodedb/src/control/server/dispatch_utils/minted/owned.rs +++ b/nodedb/src/control/server/dispatch_utils/minted/owned.rs @@ -36,8 +36,9 @@ pub(crate) struct OwnedWait { /// The key a final refusal's abort marker carries, `0` when this write's /// refusals are never final. pub final_refusal_key: u64, - /// The instant the caller stops waiting. - pub deadline: Instant, + /// The instant the caller stops waiting. `None` waits for the final + /// response: a committed write, whose core never expires it. + pub deadline: Option, pub collect: Collect, /// Where the final response goes when it arrives after the caller /// stopped waiting. @@ -125,11 +126,15 @@ async fn wait_and_close( let _ = late.send(response); } }; - let until = tokio::time::Instant::from_std(deadline); let collected = match collect { Collect::Merged { max_result_bytes } => { - tokio::time::timeout_at(until, collect_bounded_response(&mut rx, max_result_bytes)) - .await + let collect = collect_bounded_response(&mut rx, max_result_bytes); + match deadline { + Some(deadline) => { + tokio::time::timeout_at(tokio::time::Instant::from_std(deadline), collect).await + } + None => Ok(collect.await), + } } }; match collected { @@ -226,7 +231,7 @@ mod tests { wal: Arc::clone(wal), owner: owner(), final_refusal_key: 0, - deadline, + deadline: Some(deadline), collect: Collect::Merged { max_result_bytes: 1 << 20, }, @@ -264,6 +269,37 @@ mod tests { assert_eq!(late.status, Status::Ok); } + /// A committed write carries no deadline: its core never expires it, so + /// the wait reports the answer that arrives, however late, and never + /// `DeadlineExceeded`. + #[tokio::test] + async fn a_committed_write_waits_for_its_answer() { + let dir = tempfile::tempdir().expect("tempdir"); + let wal = Arc::new(WalManager::open_for_testing(&dir.path().join("wal")).expect("wal")); + let floor = OutcomeFloor::new(); + let tracker = RequestTracker::new(); + let rx = tracker.register(RequestId::new(4)); + let (minted, _lsn) = minted_record(&wal, &floor); + let report = spawn_owned_wait( + OwnedWait { + deadline: None, + ..wait(&wal, Instant::now()) + }, + rx, + minted, + ); + tokio::time::sleep(Duration::from_millis(50)).await; + let mut ok = refusal(4); + ok.status = Status::Ok; + ok.error_code = None; + assert!(tracker.complete(ok)); + let outcome = report.recv().await.expect("report"); + assert!( + matches!(&outcome, OwnedResponse::Answered { response, .. } if response.status == Status::Ok), + "a committed write reports its real answer" + ); + } + /// The caller future is dropped while it waits. The task still closes /// the records from the refusal that arrives afterwards. #[tokio::test] diff --git a/nodedb/src/control/server/dispatch_utils/submit_write/funnel/answer.rs b/nodedb/src/control/server/dispatch_utils/submit_write/funnel/answer.rs index 4cf1550e5..94edc6ba0 100644 --- a/nodedb/src/control/server/dispatch_utils/submit_write/funnel/answer.rs +++ b/nodedb/src/control/server/dispatch_utils/submit_write/funnel/answer.rs @@ -11,11 +11,12 @@ use crate::control::server::dispatch_utils::minted::{OwnedReport, OwnedResponse} /// A dispatched write's answer, before anyone waits for it. pub(super) enum Answer { /// The write carries no records. The waiter reads the channel, up to - /// `deadline` and under the byte budget. The waiter drops the receiver - /// once it stops reading, which removes the request's tracker entry. + /// `deadline` when it has one, and under the byte budget. The waiter drops + /// the receiver once it stops reading, which removes the request's tracker + /// entry. Unminted { rx: ResponseReceiver, - deadline: Instant, + deadline: Option, max_result_bytes: usize, }, /// A spawned task owns the write's records, closes them from the final @@ -32,11 +33,14 @@ impl Answer { deadline, max_result_bytes, } => { - let collected = tokio::time::timeout_at( - tokio::time::Instant::from_std(deadline), - collect_bounded_response(&mut rx, max_result_bytes), - ) - .await; + let collect = collect_bounded_response(&mut rx, max_result_bytes); + let collected = match deadline { + Some(deadline) => { + tokio::time::timeout_at(tokio::time::Instant::from_std(deadline), collect) + .await + } + None => Ok(collect.await), + }; Ok(match collected { Ok(Ok(response)) => OwnedResponse::Answered { response: Box::new(response), diff --git a/nodedb/src/control/server/dispatch_utils/submit_write/funnel/dispatch.rs b/nodedb/src/control/server/dispatch_utils/submit_write/funnel/dispatch.rs index 8aac52cbe..cfa7a64b2 100644 --- a/nodedb/src/control/server/dispatch_utils/submit_write/funnel/dispatch.rs +++ b/nodedb/src/control/server/dispatch_utils/submit_write/funnel/dispatch.rs @@ -161,8 +161,10 @@ pub(super) async fn dispatch_to_data_plane( } /// Dispatch `request`, waiting for freed capacity after each capacity -/// refusal, until the request's deadline. Returns the last refusal once the -/// deadline passes. +/// refusal, until the request's execution deadline. Returns the last refusal +/// once the deadline passes. A committed write has no execution deadline and +/// waits until capacity frees: a committed entry that fails for local load +/// leaves this replica without a write every other replica applied. /// /// A caller that fans one statement out into more requests than the tenant's /// in-flight cap dispatches each through here: the fan-out then advances as @@ -172,7 +174,9 @@ pub(crate) async fn dispatch_when_capacity_frees( request: Request, journal: Option, ) -> crate::Result<()> { - let deadline = tokio::time::Instant::from_std(request.deadline); + let deadline = request + .execution_deadline() + .map(tokio::time::Instant::from_std); let capacity_freed = match shared.dispatcher.lock() { Ok(d) => d.capacity_freed(), Err(poisoned) => poisoned.into_inner().capacity_freed(), @@ -197,8 +201,13 @@ pub(crate) async fn dispatch_when_capacity_frees( if !matches!(refusal.error, crate::Error::DispatchCapacity { .. }) { return Err(refusal.error); } - if tokio::time::timeout_at(deadline, freed).await.is_err() { - return Err(refusal.error); + match deadline { + Some(deadline) => { + if tokio::time::timeout_at(deadline, freed).await.is_err() { + return Err(refusal.error); + } + } + None => freed.await, } request = refusal.request; } diff --git a/nodedb/src/control/server/dispatch_utils/submit_write/funnel/driver.rs b/nodedb/src/control/server/dispatch_utils/submit_write/funnel/driver.rs index d89e69ef0..5d7bd2394 100644 --- a/nodedb/src/control/server/dispatch_utils/submit_write/funnel/driver.rs +++ b/nodedb/src/control/server/dispatch_utils/submit_write/funnel/driver.rs @@ -170,6 +170,7 @@ pub(crate) async fn enqueue_write( // rather than failing: a committed entry that fails for local load leaves // this replica without a write every other replica applied. let waits_for_capacity = matches!(ordering, WriteOrdering::AlreadyOrdered); + let collect_until = (!waits_for_capacity).then_some(deadline); // A committed entry carries the rows its proposer resolved. Resolving it // again here will let this replica accept other lines than its peers. if waits_for_capacity && crate::control::write_resolve::is_unresolved_ingest(&plan) { @@ -459,7 +460,7 @@ pub(crate) async fn enqueue_write( wal: Arc::clone(&shared.wal), owner, final_refusal_key, - deadline, + deadline: collect_until, collect: Collect::Merged { max_result_bytes }, late: late_tx, }, @@ -468,7 +469,7 @@ pub(crate) async fn enqueue_write( )), None => Answer::Unminted { rx: dispatch_outcome.rx, - deadline, + deadline: collect_until, max_result_bytes, }, }; @@ -479,7 +480,7 @@ pub(crate) async fn enqueue_write( request_id: dispatch_outcome.request_id, answer, max_result_bytes, - deadline, + deadline: collect_until, dispatch_started: dispatch_outcome.dispatch_started, tenant_id, database_id, diff --git a/nodedb/src/control/server/dispatch_utils/submit_write/funnel/response.rs b/nodedb/src/control/server/dispatch_utils/submit_write/funnel/response.rs index 0deadbc37..285d1dc22 100644 --- a/nodedb/src/control/server/dispatch_utils/submit_write/funnel/response.rs +++ b/nodedb/src/control/server/dispatch_utils/submit_write/funnel/response.rs @@ -25,7 +25,9 @@ pub(super) struct ResponsePhaseInput { pub answer: Answer, /// The byte budget of the merged response. pub max_result_bytes: usize, - pub deadline: Instant, + /// The instant the caller stops waiting. `None` for a committed write, + /// whose core never expires it. + pub deadline: Option, pub dispatch_started: Instant, pub tenant_id: TenantId, pub database_id: DatabaseId, @@ -159,7 +161,7 @@ pub(super) async fn collect_classify_and_finish( // A producer that stopped after the deadline stopped because the // statement ran out of time. Reporting the closure will hand the // client an internal error for its own timeout. - if std::time::Instant::now() >= deadline { + if deadline.is_some_and(|deadline| std::time::Instant::now() >= deadline) { return Err(crate::Error::DeadlineExceeded { request_id }); } return Err(crate::Error::Dispatch { diff --git a/nodedb/src/control/server/pgwire/handler/prepared/parser.rs b/nodedb/src/control/server/pgwire/handler/prepared/parser.rs index 7958ee328..a804c225e 100644 --- a/nodedb/src/control/server/pgwire/handler/prepared/parser.rs +++ b/nodedb/src/control/server/pgwire/handler/prepared/parser.rs @@ -250,6 +250,8 @@ impl NodeDbQueryParser { tenant_id: u64, database_id: crate::types::DatabaseId, ) -> crate::control::planner::catalog_adapter::OriginCatalog { + // Describe plans the statement without executing it: sequence + // functions resolve to their type, and no counter advances. crate::control::planner::catalog_adapter::OriginCatalog::new( Arc::clone(&self.state.credentials), self.state.array_catalog.clone(), @@ -257,6 +259,8 @@ impl NodeDbQueryParser { database_id, Some(Arc::clone(&self.state.retention_policy_registry)), ) + .with_sequence_registry(Arc::clone(&self.state.sequence_registry)) + .describing() } /// Infer parameter and result types using the nodedb-sql catalog. diff --git a/nodedb/src/control/server/shared/session/ddl_flush.rs b/nodedb/src/control/server/shared/session/ddl_flush.rs index 58c1589f9..1f000316f 100644 --- a/nodedb/src/control/server/shared/session/ddl_flush.rs +++ b/nodedb/src/control/server/shared/session/ddl_flush.rs @@ -17,6 +17,7 @@ use nodedb_cluster::{MetadataEntry, PendingDdlObject, encode_entry}; use crate::control::catalog_entry::{self, CatalogEntry}; use crate::control::constraint_install::{ConstraintTarget, constraint_targets}; +use crate::control::lease::DrainHolds; use crate::control::metadata_proposer::{DdlPrepareLease, MetadataRaftHandle}; use crate::control::security::catalog::SystemCatalog; use crate::control::state::SharedState; @@ -183,8 +184,12 @@ fn committed_before_image( /// Reserve `buffered` under a fresh fencing token via `DdlPendingPropose`, /// without dispatching any DML. Takes the preparation lock and lease, drains /// and stamps the batch, then reserves each stamped statement as a -/// [`PendingDdlObject`] instead of applying it. A failure releases the lease -/// before it returns. +/// [`PendingDdlObject`] instead of applying it. +/// +/// A failure ends every drain it started, cancels any pending record its +/// propose left, and releases the lease before it returns. On success the +/// record owns the drains: its finalize's catalog entries end them, and so +/// does its cancel. async fn propose_pending_buffered<'a>( state: &'a SharedState, sessions: &SessionStore, @@ -196,12 +201,14 @@ async fn propose_pending_buffered<'a>( let lease = crate::control::metadata_proposer::acquire_ddl_prepare_lease_async(state, handle.as_ref()) .await?; + let mut drains = DrainHolds::new(state); let reserved = reserve_buffered( state, sessions, session_id, handle.as_ref(), lease.token(), + &mut drains, buffered, ) .await; @@ -210,15 +217,19 @@ async fn propose_pending_buffered<'a>( log_index, objects, constraint_targets, - }) => Ok(PendingDdlHandle { - token: lease.token(), - log_index, - lease, - objects, - constraint_targets, - }), + }) => { + drains.keep(); + Ok(PendingDdlHandle { + token: lease.token(), + log_index, + lease, + objects, + constraint_targets, + }) + } Err(error) => { - lease.release().await; + drains.end().await; + lease.abandon().await; Err(error) } } @@ -234,13 +245,15 @@ struct Reserved { /// Drain, stamp, and reserve `buffered` under the preparation lease `token`. /// Returns the reservation's log index, its reserved objects, and the -/// collections whose constraint version it changes. +/// collections whose constraint version it changes. Every drain it starts is +/// recorded in `drains`, for the caller to keep or end. async fn reserve_buffered( state: &SharedState, sessions: &SessionStore, session_id: SessionId, handle: &dyn MetadataRaftHandle, token: u64, + drains: &mut DrainHolds<'_>, buffered: DdlBuffer, ) -> crate::Result { // Checked under the preparation lease, which every DDL takes: no other @@ -262,6 +275,7 @@ async fn reserve_buffered( // conflict with itself. let own_holds = sessions.own_lease_hold_count(session_id, &descriptor_id, prior_version); + drains.hold(descriptor_id.clone(), nodedb_cluster::DrainOwner::Ddl); crate::control::lease::drain_for_ddl_async( state, descriptor_id, @@ -346,7 +360,13 @@ pub(super) async fn finalize_pending( } = handle; tracing::debug!(token, log_index, "finalizing pending DDL"); let finalized = finalize_reserved(state, token, &objects).await; - lease.release().await; + // A finalize whose outcome is unknown can still apply, or never apply and + // leave its record. The abandon's cancel follows it in the log, so it is a + // no-op after an applied finalize and resolves the record otherwise. + match &finalized { + Ok(_) => lease.release().await, + Err(_) => lease.abandon().await, + } let (log_index, bears_authorization) = finalized?; // A changed constraint set is live on the collection's data log before // COMMIT returns, so the first CRDT apply after it is not fenced. The @@ -728,6 +748,83 @@ mod tests { cluster.shutdown().await; } + /// A reserved COMMIT that never finalizes leaves nothing behind: its + /// pending record is cancelled, the drain its alter started ends, and the + /// preparation lease is free again, without a restart. + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + async fn an_unfinalized_commit_cancels_its_record_and_ends_its_drains() { + let cluster = test_one_node::boot().await; + { + let state = &cluster.state; + let mut existing = + crate::control::security::catalog::StoredCollection::stamped_for_test( + 7, "orders", "alice", + ); + existing.descriptor_version = 3; + state + .credentials + .catalog() + .put_collection(nodedb_types::DatabaseId::DEFAULT, &existing) + .expect("store the collection the commit alters"); + let mut altered = existing.clone(); + altered.owner = "bob".into(); + + let (sessions, session_id) = test_session(); + let plan = conn_scope::scoped(async { + ddl_buffer::activate(); + assert!(ddl_buffer::try_buffer(CatalogEntry::PutCollection( + Box::new(altered) + ))); + begin_commit(state, &sessions, session_id).await + }) + .await + .expect("begin_commit must succeed"); + let handle = match plan { + DdlCommitPlan::Pending(handle) => handle, + DdlCommitPlan::None => panic!("a buffered alter must reserve a pending record"), + }; + let token = handle.token; + let descriptor = nodedb_cluster::DescriptorId::new( + 0, + 7, + nodedb_cluster::DescriptorKind::Collection, + "orders", + ); + assert!(state.pending_ddl.contains(token)); + assert!( + state.lease_drain.is_draining(&descriptor, 3), + "the reservation drains the altered collection" + ); + + drop(handle); + + let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(5); + loop { + let owner_free = state + .metadata_ddl + .owner + .lock() + .expect("owner lock") + .is_none(); + if !state.pending_ddl.contains(token) + && !state.lease_drain.is_draining(&descriptor, 3) + && owner_free + { + break; + } + assert!( + tokio::time::Instant::now() < deadline, + "an unfinalized commit left state behind: record={} draining={} lease_held={}", + state.pending_ddl.contains(token), + state.lease_drain.is_draining(&descriptor, 3), + !owner_free + ); + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + } + } + cluster.shutdown().await; + } + #[test] fn reverse_create_purges_a_created_collection() { let stored = crate::control::security::catalog::StoredCollection::new(7, "orders", "alice"); diff --git a/nodedb/src/control/startup/phase.rs b/nodedb/src/control/startup/phase.rs index 321ee5018..31ccc8a28 100644 --- a/nodedb/src/control/startup/phase.rs +++ b/nodedb/src/control/startup/phase.rs @@ -15,7 +15,7 @@ use std::fmt; /// Total number of phases. Kept in sync with the enum below by /// the `phase_order_matches_u8` unit test. -pub const PHASE_COUNT: usize = 13; +pub const PHASE_COUNT: usize = 14; /// Startup phase. Ordered — use `Ord` / `PartialOrd` to compare. #[derive(Copy, Clone, Debug, Eq, PartialEq, Ord, PartialOrd, Hash)] @@ -27,38 +27,42 @@ pub enum StartupPhase { WalRecovery = 1, /// `SystemCatalog` (redb) opened, credential store loaded. ClusterCatalogOpen = 2, + /// Every Data Plane core finished its WAL replay and drains its request + /// ring. A request sent to a core before this waits in the ring, so boot + /// work that dispatches to the cores waits for this phase first. + DataPlaneReplay = 3, /// Metadata raft group's committed log has been applied up /// to the last-observed index. `MetadataCache.applied_index` /// is current. - RaftMetadataReplay = 3, + RaftMetadataReplay = 4, /// Post-apply hooks rebuilt every in-memory registry /// (triggers, streams, schedules, permissions, etc.) from /// the now-fresh redb state. - SchemaCacheWarmup = 4, + SchemaCacheWarmup = 5, /// Applied-index gate, redb cross-table integrity, and /// in-memory registry ⇔ redb verification have all run /// without raising unrepairable divergences. See /// `control::cluster::recovery_check`. - CatalogSanityCheck = 5, + CatalogSanityCheck = 6, /// All data raft groups (vShards hosting data) have caught /// up to their committed watermark. - DataGroupsReplay = 6, + DataGroupsReplay = 7, /// Listener sockets bound (pgwire / HTTP / ILP / RESP / /// native). Not yet accepting requests. - TransportBind = 7, + TransportBind = 8, /// Parallel dials completed against every known peer so /// the QUIC peer cache is hot before any replicated /// request fires. - WarmPeers = 8, + WarmPeers = 9, /// Health monitor running. - HealthLoopStart = 9, + HealthLoopStart = 10, /// Cluster readiness completed: the node holds its first authorization /// lease and its gateway is installed. Listener accept loops may process /// requests once they exist. The client protocols do not listen yet. - GatewayEnable = 10, + GatewayEnable = 11, /// The client protocols listen and every HTTP route answers. The final /// phase: the node is ready, and `/healthz` reports `ok`. - Serving = 11, + Serving = 12, /// Terminal state — entered via [`StartupSequencer::fail`] or /// when a [`ReadyGate`] is dropped without firing. All /// [`StartupGate::await_phase`] waiters wake with an error. @@ -66,7 +70,7 @@ pub enum StartupPhase { /// [`StartupSequencer::fail`]: super::startup_sequencer::StartupSequencer::fail /// [`ReadyGate`]: super::gate::ReadyGate /// [`StartupGate::await_phase`]: super::gate::StartupGate::await_phase - Failed = 12, + Failed = 13, } impl StartupPhase { @@ -77,6 +81,7 @@ impl StartupPhase { Self::Boot => "boot", Self::WalRecovery => "wal_recovery", Self::ClusterCatalogOpen => "cluster_catalog_open", + Self::DataPlaneReplay => "data_plane_replay", Self::RaftMetadataReplay => "raft_metadata_replay", Self::SchemaCacheWarmup => "schema_cache_warmup", Self::CatalogSanityCheck => "catalog_sanity_check", @@ -95,7 +100,8 @@ impl StartupPhase { match self { Self::Boot => Some(Self::WalRecovery), Self::WalRecovery => Some(Self::ClusterCatalogOpen), - Self::ClusterCatalogOpen => Some(Self::RaftMetadataReplay), + Self::ClusterCatalogOpen => Some(Self::DataPlaneReplay), + Self::DataPlaneReplay => Some(Self::RaftMetadataReplay), Self::RaftMetadataReplay => Some(Self::SchemaCacheWarmup), Self::SchemaCacheWarmup => Some(Self::CatalogSanityCheck), Self::CatalogSanityCheck => Some(Self::DataGroupsReplay), @@ -116,16 +122,17 @@ impl StartupPhase { 0 => Some(Self::Boot), 1 => Some(Self::WalRecovery), 2 => Some(Self::ClusterCatalogOpen), - 3 => Some(Self::RaftMetadataReplay), - 4 => Some(Self::SchemaCacheWarmup), - 5 => Some(Self::CatalogSanityCheck), - 6 => Some(Self::DataGroupsReplay), - 7 => Some(Self::TransportBind), - 8 => Some(Self::WarmPeers), - 9 => Some(Self::HealthLoopStart), - 10 => Some(Self::GatewayEnable), - 11 => Some(Self::Serving), - 12 => Some(Self::Failed), + 3 => Some(Self::DataPlaneReplay), + 4 => Some(Self::RaftMetadataReplay), + 5 => Some(Self::SchemaCacheWarmup), + 6 => Some(Self::CatalogSanityCheck), + 7 => Some(Self::DataGroupsReplay), + 8 => Some(Self::TransportBind), + 9 => Some(Self::WarmPeers), + 10 => Some(Self::HealthLoopStart), + 11 => Some(Self::GatewayEnable), + 12 => Some(Self::Serving), + 13 => Some(Self::Failed), _ => None, } } @@ -154,6 +161,7 @@ mod tests { StartupPhase::Boot, StartupPhase::WalRecovery, StartupPhase::ClusterCatalogOpen, + StartupPhase::DataPlaneReplay, StartupPhase::RaftMetadataReplay, StartupPhase::SchemaCacheWarmup, StartupPhase::CatalogSanityCheck, diff --git a/nodedb/src/control/startup/startup_sequencer.rs b/nodedb/src/control/startup/startup_sequencer.rs index e358aa01b..4a09acf93 100644 --- a/nodedb/src/control/startup/startup_sequencer.rs +++ b/nodedb/src/control/startup/startup_sequencer.rs @@ -353,7 +353,9 @@ mod tests { assert_eq!(gate.current_phase(), StartupPhase::WalRecovery); g2.fire(); - assert_eq!(gate.current_phase(), StartupPhase::ClusterCatalogOpen); + // `DataPlaneReplay` has no gate here, so the sequencer passes it and + // stops before `RaftMetadataReplay`, whose gate is pending. + assert_eq!(gate.current_phase(), StartupPhase::DataPlaneReplay); g3.fire(); // After g3 fires, sequencer advances to RaftMetadataReplay and then @@ -533,6 +535,7 @@ mod tests { StartupPhase::Boot => StartupPhase::Boot.next(), StartupPhase::WalRecovery => StartupPhase::WalRecovery.next(), StartupPhase::ClusterCatalogOpen => StartupPhase::ClusterCatalogOpen.next(), + StartupPhase::DataPlaneReplay => StartupPhase::DataPlaneReplay.next(), StartupPhase::RaftMetadataReplay => StartupPhase::RaftMetadataReplay.next(), StartupPhase::SchemaCacheWarmup => StartupPhase::SchemaCacheWarmup.next(), StartupPhase::CatalogSanityCheck => StartupPhase::CatalogSanityCheck.next(), diff --git a/nodedb/src/data/executor/columnar_checkpoint/format.rs b/nodedb/src/data/executor/columnar_checkpoint/format.rs index 3a5f303a3..e8f37fe9c 100644 --- a/nodedb/src/data/executor/columnar_checkpoint/format.rs +++ b/nodedb/src/data/executor/columnar_checkpoint/format.rs @@ -40,6 +40,18 @@ pub(crate) struct ColumnarCheckpointManifest { pub replay: ReplayStamp, } +/// Encode a manifest publishing `generation`, for tests that need a live +/// generation on disk without driving a whole core through a checkpoint. +#[cfg(test)] +pub(crate) fn test_manifest_bytes(generation: u64) -> Vec { + zerompk::to_msgpack_vec(&ColumnarCheckpointManifest { + format_version: COLUMNAR_CKPT_FORMAT_VERSION, + generation, + replay: ReplayStamp::through(0), + }) + .expect("manifest encode is infallible for this fixed struct") +} + /// One collection's full engine state within a generation. #[derive( Debug, Clone, Serialize, Deserialize, zerompk::ToMessagePack, zerompk::FromMessagePack, diff --git a/nodedb/src/data/executor/columnar_checkpoint/manifest.rs b/nodedb/src/data/executor/columnar_checkpoint/manifest.rs index 0fbd39517..f507f025d 100644 --- a/nodedb/src/data/executor/columnar_checkpoint/manifest.rs +++ b/nodedb/src/data/executor/columnar_checkpoint/manifest.rs @@ -10,49 +10,62 @@ use crate::data::executor::core_loop::CoreLoop; impl CoreLoop { /// Read the live manifest, or `Ok(None)` when the file is genuinely absent. - /// - /// A present-but-unreadable, undecodable, or version-mismatched manifest is - /// `Err`, not `None`: the manifest is the only record of the LSN the - /// generation it names is durable through, and treating corruption as - /// "no generation" would let the caller install no floor while the WAL - /// below that LSN may already be gone. + /// See [`read_columnar_manifest_at`]. pub(in crate::data::executor) fn read_columnar_manifest( &self, ckpt_dir: &std::path::Path, ) -> Result, CheckpointDecodeError> { - let path = ckpt_dir.join(COLUMNAR_CKPT_MANIFEST); - if !path.exists() { - return Ok(None); + read_columnar_manifest_at(ckpt_dir) + } +} + +/// Read the live manifest in `ckpt_dir`, or `Ok(None)` when the file is +/// genuinely absent. +/// +/// A present-but-unreadable, undecodable, or version-mismatched manifest is +/// `Err`, not `None`: the manifest is the only record of the LSN the +/// generation it names is durable through, and treating corruption as +/// "no generation" would let the caller install no floor while the WAL +/// below that LSN may already be gone. +/// +/// A free function because reclaim, which runs against a data dir and not a +/// live core, resolves the live generation through this same reader, and the +/// two must never diverge on what "live" means. +pub(crate) fn read_columnar_manifest_at( + ckpt_dir: &std::path::Path, +) -> Result, CheckpointDecodeError> { + let path = ckpt_dir.join(COLUMNAR_CKPT_MANIFEST); + if !path.exists() { + return Ok(None); + } + let bytes = nodedb_wal::segment::read_checkpoint_framed(&path).map_err(|source| { + CheckpointDecodeError::ReadFile { + path: path.clone(), + source, } - let bytes = nodedb_wal::segment::read_checkpoint_framed(&path).map_err(|source| { - CheckpointDecodeError::ReadFile { + })?; + let manifest = + zerompk::from_msgpack::(&bytes).map_err(|source| { + CheckpointDecodeError::MsgpackDecode { path: path.clone(), source, } })?; - let manifest = - zerompk::from_msgpack::(&bytes).map_err(|source| { - CheckpointDecodeError::MsgpackDecode { - path: path.clone(), - source, - } - })?; - if manifest.format_version != COLUMNAR_CKPT_FORMAT_VERSION { - return Err(CheckpointDecodeError::FormatVersion { - path: path.clone(), - found: manifest.format_version, - expected: COLUMNAR_CKPT_FORMAT_VERSION, - }); - } - manifest - .replay - .validate() - .map_err(|source| CheckpointDecodeError::InvalidReplayStamp { - path: path.clone(), - source, - })?; - Ok(Some(manifest)) + if manifest.format_version != COLUMNAR_CKPT_FORMAT_VERSION { + return Err(CheckpointDecodeError::FormatVersion { + path: path.clone(), + found: manifest.format_version, + expected: COLUMNAR_CKPT_FORMAT_VERSION, + }); } + manifest + .replay + .validate() + .map_err(|source| CheckpointDecodeError::InvalidReplayStamp { + path: path.clone(), + source, + })?; + Ok(Some(manifest)) } /// Wrap a filesystem failure as the columnar engine's typed storage error. diff --git a/nodedb/src/data/executor/columnar_checkpoint/mod.rs b/nodedb/src/data/executor/columnar_checkpoint/mod.rs index e1db44cd7..56deb3dc6 100644 --- a/nodedb/src/data/executor/columnar_checkpoint/mod.rs +++ b/nodedb/src/data/executor/columnar_checkpoint/mod.rs @@ -124,4 +124,9 @@ mod manifest; mod paths; mod write; -pub(crate) use paths::{COLUMNAR_CKPT_MANIFEST, columnar_ckpt_dir, columnar_ckpt_gen_dir}; +#[cfg(test)] +pub(crate) use format::test_manifest_bytes; +pub(crate) use manifest::read_columnar_manifest_at; +pub(crate) use paths::{ + COLUMNAR_CKPT_MANIFEST, columnar_ckpt_dir, columnar_ckpt_filename, columnar_ckpt_gen_dir, +}; diff --git a/nodedb/src/data/executor/handlers/reclaim/columnar.rs b/nodedb/src/data/executor/handlers/reclaim/columnar.rs new file mode 100644 index 000000000..c30f5a9b0 --- /dev/null +++ b/nodedb/src/data/executor/handlers/reclaim/columnar.rs @@ -0,0 +1,227 @@ +// SPDX-License-Identifier: BUSL-1.1 + +//! Columnar engine reclaim — unlink a dropped collection's columnar +//! checkpoint file. +//! +//! Checkpoint layout: +//! `{data_dir}/columnar-ckpt/core-{n}/gen-{m}/db-{did}-tenant-{tid}-coll-{hex(coll)}.ckpt`, +//! with each core's `MANIFEST` naming its live generation. Reclaim walks EVERY +//! core's directory, because `data_dir` is shared and the dropped collection +//! can be routed to any of them. Reclaim resolves each core's live +//! generation through the same manifest reader the load path uses so the two +//! can never disagree about which files are reachable. The filename is built +//! by the SAME helper the write path uses ([`columnar_ckpt_filename`]). +//! +//! Every columnar-family collection has this file: `columnar`, `timeseries` +//! on columnar, and `spatial`, whose rows the columnar engine stores. The +//! columnar restore reinstalls the engine and rebuilds a spatial collection's +//! R-tree from the file's rows, so a file left behind resurrects a dropped +//! collection on the next boot, and the next checkpoint makes it permanent. +//! +//! Unlinking a file out of a published generation does not break the +//! manifest's promise. That promise is about the collections that are LIVE: +//! the dropped collection's records above the replay floor are tombstoned, and +//! the ones below it are no longer in the WAL. + +use std::path::Path; + +use tracing::debug; + +use super::{ReclaimError, ReclaimStats, Result}; +use crate::data::executor::columnar_checkpoint::{ + columnar_ckpt_filename, columnar_ckpt_gen_dir, read_columnar_manifest_at, +}; + +/// Unlink the columnar checkpoint file for `(database_id, tenant_id, +/// collection)` from every core's live generation. Returns stats; idempotent. +pub fn reclaim_columnar_checkpoints( + data_dir: &Path, + database_id: u64, + tenant_id: u64, + collection: &str, +) -> Result { + let root = data_dir.join("columnar-ckpt"); + let cores = match std::fs::read_dir(&root) { + Ok(entries) => entries, + Err(source) if source.kind() == std::io::ErrorKind::NotFound => { + return Ok(ReclaimStats::default()); + } + Err(source) => { + return Err(ReclaimError::Io { + operation: "read columnar checkpoint root", + path: root, + source, + }); + } + }; + + let file_name = columnar_ckpt_filename(database_id, tenant_id, collection); + let tmp_name = format!("{file_name}.tmp"); + + let mut stats = ReclaimStats::default(); + for core_entry in cores { + let core_entry = core_entry.map_err(|source| ReclaimError::Io { + operation: "read columnar checkpoint core entry", + path: root.clone(), + source, + })?; + let core_dir = core_entry.path(); + if !core_dir.is_dir() { + continue; + } + // No manifest means no reachable generation for this core, so there is + // nothing a boot could restore and nothing to reclaim. A corrupt + // manifest is fail-closed: skipping it could release a same-name CREATE + // while predecessor files remain reachable. + let gen_dir = match read_columnar_manifest_at(&core_dir) { + Ok(Some(manifest)) => columnar_ckpt_gen_dir(&core_dir, manifest.generation), + Ok(None) => continue, + Err(error) => { + return Err(ReclaimError::Manifest { + engine: "columnar", + path: core_dir.join("MANIFEST"), + detail: error.to_string(), + }); + } + }; + for name in [&file_name, &tmp_name] { + unlink(&gen_dir.join(name), &mut stats)?; + } + } + Ok(stats) +} + +/// Unlink `path`. A missing file is a no-op, not an error. +fn unlink(path: &Path, stats: &mut ReclaimStats) -> Result<()> { + let size = match std::fs::metadata(path) { + Ok(metadata) => metadata.len(), + Err(source) if source.kind() == std::io::ErrorKind::NotFound => return Ok(()), + Err(source) => { + return Err(ReclaimError::Io { + operation: "stat columnar checkpoint", + path: path.to_path_buf(), + source, + }); + } + }; + match std::fs::remove_file(path) { + Ok(()) => { + stats.files_unlinked = stats.files_unlinked.saturating_add(1); + stats.bytes_freed = stats.bytes_freed.saturating_add(size); + debug!(path = %path.display(), size, "columnar reclaim: unlinked"); + Ok(()) + } + Err(source) if source.kind() == std::io::ErrorKind::NotFound => Ok(()), + Err(source) => Err(ReclaimError::Io { + operation: "unlink columnar checkpoint", + path: path.to_path_buf(), + source, + }), + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::data::executor::columnar_checkpoint::{columnar_ckpt_dir, test_manifest_bytes}; + use tempfile::TempDir; + + /// Publish a live generation under `core_id` holding `files`, through the + /// same manifest the load path reads. + fn publish(data_dir: &Path, core_id: usize, generation: u64, files: &[(&str, &[u8])]) { + let ckpt_dir = columnar_ckpt_dir(data_dir, core_id); + let gen_dir = columnar_ckpt_gen_dir(&ckpt_dir, generation); + std::fs::create_dir_all(&gen_dir).expect("mkdir"); + for (name, bytes) in files { + std::fs::write(gen_dir.join(name), bytes).expect("write"); + } + nodedb_wal::segment::write_checkpoint_framed( + &ckpt_dir, + "MANIFEST", + &test_manifest_bytes(generation), + ) + .expect("manifest"); + } + + fn live_file(data_dir: &Path, core_id: usize, generation: u64, name: &str) -> bool { + columnar_ckpt_gen_dir(&columnar_ckpt_dir(data_dir, core_id), generation) + .join(name) + .exists() + } + + /// The dropped collection's file goes from every core's live generation; + /// another collection's file stays. + #[test] + fn unlinks_the_collection_across_cores() { + let tmp = TempDir::new().expect("tempdir"); + let base = tmp.path(); + let dropped = columnar_ckpt_filename(0, 1, "places"); + let kept = columnar_ckpt_filename(0, 1, "places_archive"); + publish(base, 0, 3, &[(&dropped, b"x"), (&kept, b"keep")]); + publish(base, 1, 0, &[(&dropped, b"yy")]); + + let stats = reclaim_columnar_checkpoints(base, 0, 1, "places").expect("reclaim"); + assert_eq!(stats.files_unlinked, 2); + assert_eq!(stats.bytes_freed, 3); + assert!(!live_file(base, 0, 3, &dropped)); + assert!(!live_file(base, 1, 0, &dropped)); + assert!( + live_file(base, 0, 3, &kept), + "a collection whose name extends the dropped one survives" + ); + } + + /// Another database's or tenant's same-name collection survives. + #[test] + fn leaves_other_scopes_alone() { + let tmp = TempDir::new().expect("tempdir"); + let base = tmp.path(); + let other_db = columnar_ckpt_filename(1024, 1, "places"); + let other_tenant = columnar_ckpt_filename(0, 2, "places"); + publish(base, 0, 1, &[(&other_db, b"a"), (&other_tenant, b"b")]); + + let stats = reclaim_columnar_checkpoints(base, 0, 1, "places").expect("reclaim"); + assert_eq!(stats.files_unlinked, 0); + assert!(live_file(base, 0, 1, &other_db)); + assert!(live_file(base, 0, 1, &other_tenant)); + } + + /// A missing root, or a core with no manifest, has nothing to reclaim. + #[test] + fn absent_root_or_manifest_is_a_no_op() { + let tmp = TempDir::new().expect("tempdir"); + let base = tmp.path(); + assert_eq!( + reclaim_columnar_checkpoints(base, 0, 1, "places") + .expect("absent root") + .files_unlinked, + 0 + ); + std::fs::create_dir_all(columnar_ckpt_dir(base, 0)).expect("mkdir"); + assert_eq!( + reclaim_columnar_checkpoints(base, 0, 1, "places") + .expect("absent manifest") + .files_unlinked, + 0 + ); + } + + /// A corrupt manifest is an error, never a silent skip. + #[test] + fn corrupt_manifest_fails_closed() { + let tmp = TempDir::new().expect("tempdir"); + let base = tmp.path(); + let ckpt_dir = columnar_ckpt_dir(base, 0); + std::fs::create_dir_all(&ckpt_dir).expect("mkdir"); + std::fs::write(ckpt_dir.join("MANIFEST"), b"not a manifest").expect("write"); + let error = reclaim_columnar_checkpoints(base, 0, 1, "places") + .expect_err("a corrupt manifest must fail the reclaim"); + assert!(matches!( + error, + ReclaimError::Manifest { + engine: "columnar", + .. + } + )); + } +} diff --git a/nodedb/src/data/executor/handlers/reclaim/mod.rs b/nodedb/src/data/executor/handlers/reclaim/mod.rs index 3a61ec9e8..f725e974a 100644 --- a/nodedb/src/data/executor/handlers/reclaim/mod.rs +++ b/nodedb/src/data/executor/handlers/reclaim/mod.rs @@ -8,6 +8,11 @@ //! been evicted but before the JSON summary is built, so the handler //! picks up per-file byte counts for the `bytes_reclaimed` metric. //! +//! The columnar file is reclaimed for every columnar-family collection +//! (`columnar`, `timeseries` on columnar, `spatial`): a `spatial` collection's +//! rows live in the columnar engine, and the columnar restore rebuilds its +//! R-tree from them. +//! //! Engines whose persistent state is shared-redb (document, //! document-strict, FTS, graph edges) or in-memory only (the KV hash //! index) are documented inline in the parent handler — no separate @@ -15,6 +20,7 @@ //! write per-collection checkpoint or partition files under //! `{data_dir}/...`. +pub mod columnar; pub mod crdt; mod error; pub mod sparse_vector; diff --git a/nodedb/src/data/executor/handlers/unregister_collection.rs b/nodedb/src/data/executor/handlers/unregister_collection.rs index 1b45c828a..bf24046fe 100644 --- a/nodedb/src/data/executor/handlers/unregister_collection.rs +++ b/nodedb/src/data/executor/handlers/unregister_collection.rs @@ -372,6 +372,19 @@ impl CoreLoop { ) }, )?); + l1.merge(retry_reclaim( + "columnar checkpoints", + tid_raw, + collection, + || { + reclaim::columnar::reclaim_columnar_checkpoints( + &self.data_dir, + db_raw, + tid_raw, + collection, + ) + }, + )?); l1.merge(retry_reclaim( "spatial checkpoints", tid_raw, diff --git a/nodedb/src/diag/context/lease.rs b/nodedb/src/diag/context/lease.rs index 47f5d36ed..4123d7aca 100644 --- a/nodedb/src/diag/context/lease.rs +++ b/nodedb/src/diag/context/lease.rs @@ -49,6 +49,73 @@ impl DomainContext for DescriptorLeaseNotRenewed<'_> { } } +/// A DDL preparation lease whose failure cleanup did not apply. +pub(in crate::diag) struct DdlCleanupNotCompleted<'a> { + /// The cleanup step that failed. + pub step: &'static str, + /// The lease's fencing token. + pub token: u64, + /// What failed, without the per-occurrence detail. + pub error_class: &'a str, +} + +impl DomainContext for DdlCleanupNotCompleted<'_> { + fn domain_kind(&self) -> &'static str { + "nodedb.ddl_cleanup_not_completed" + } + + fn grouping_key(&self) -> String { + // The token is the occurrence. + format!("step={} cause={}", self.step, self.error_class) + } + + fn to_json(&self) -> Value { + json!({ + "step": self.step, + "token": self.token, + "error_class": self.error_class, + "why_fatal": "until it is cancelled and released, the lease blocks every DDL in the \ + cluster and its pending record keeps its descriptors reserved", + "operator_action": "the background releaser retries the cleanup and the metadata \ + leader reclaims the lease after its expiry; check metadata-group \ + leadership if either report repeats", + }) + } +} + +/// A descriptor drain whose holder gave up and whose end did not apply. +pub(in crate::diag) struct DescriptorDrainNotEnded<'a> { + /// Debug form of the descriptor id. + pub descriptor: &'a str, + /// Debug form of the drain owner. + pub owner: &'a str, + /// What failed, without the per-occurrence detail. + pub error_class: &'a str, +} + +impl DomainContext for DescriptorDrainNotEnded<'_> { + fn domain_kind(&self) -> &'static str { + "nodedb.descriptor_drain_not_ended" + } + + fn grouping_key(&self) -> String { + // The descriptor is the occurrence. + format!("cause={}", self.error_class) + } + + fn to_json(&self) -> Value { + json!({ + "descriptor": self.descriptor, + "owner": self.owner, + "error_class": self.error_class, + "why_fatal": "a drain has no expiry: until it ends, lease acquires on the descriptor \ + fail and plans on it are refused as a retryable schema change", + "operator_action": "the background releaser retries the end; check metadata-group \ + leadership if the report repeats", + }) + } +} + #[cfg(test)] mod tests { use super::*; diff --git a/nodedb/src/diag/context/mod.rs b/nodedb/src/diag/context/mod.rs index dd77470d7..b8d05a830 100644 --- a/nodedb/src/diag/context/mod.rs +++ b/nodedb/src/diag/context/mod.rs @@ -63,7 +63,9 @@ pub(in crate::diag) use held_entry::{ApplyPanicked, HeldEntryDebtNotRecorded, Re pub(in crate::diag) use index_rebuild::IndexRebuildNotInstalled; pub(in crate::diag) use ingest::IlpAcceptedLinesDropped; pub use ingest::IlpFlushOutcome; -pub(in crate::diag) use lease::DescriptorLeaseNotRenewed; +pub(in crate::diag) use lease::{ + DdlCleanupNotCompleted, DescriptorDrainNotEnded, DescriptorLeaseNotRenewed, +}; pub(in crate::diag) use outcome_floor::{WriteWindowHeld, WriteWindowLeaked}; pub use quota::{DATABASE_SCOPE, TENANT_SCOPE}; pub(in crate::diag) use quota::{ diff --git a/nodedb/src/diag/mod.rs b/nodedb/src/diag/mod.rs index 38661b7cb..3daae96c4 100644 --- a/nodedb/src/diag/mod.rs +++ b/nodedb/src/diag/mod.rs @@ -19,18 +19,19 @@ pub use recording::{ continuous_aggregate_state_fault, crdt_dead_letter_not_enqueued, crdt_dead_letter_not_restored, crdt_dead_letter_not_stored, custom_crs_set_not_applied, cut_barrier_not_placed, data_plane_core_fail_stopped, data_plane_response_lost, data_plane_responses_lost, - descriptor_lease_not_renewed, entry_kind, fts_index_update_failed, - held_entry_debt_not_recorded, history_compaction_not_applied, ilp_invalid_utf8_drop, - ilp_line_read_drop, index_rebuild_not_installed, metadata_apply_wedged, - orphaned_index_entry_after_delete, quota_row_invalid, quota_row_undecodable, - quota_row_write_failed, quota_scope_purge_incomplete, quota_scope_replay_aborted, - raft_entries_reapplied, raft_entry_reapplied, redo_abandon_given_up, redo_entry_malformed, - redo_snapshot_debt_not_recorded, redo_stream_lost_here, replay_record_unapplied, - retention_autowire_orphaned, scope_quota_not_installed, strict_row_image_unrendered, - strict_row_undecodable, synonym_group_not_applied, timeseries_flush_not_installed, - timeseries_idle_flush_failed, timeseries_partition_unreadable, - timeseries_row_image_undecodable, vector_build_failed, vector_builder_disconnected, - vector_builder_spawn_failed, vector_index_not_applied, vector_payload_undecodable, - vector_rebuild_unreadable, vshard_departure_purge_failed, wal_archival_failed_truncation_held, - write_acked_without_durability, write_window_held, write_window_leaked, + ddl_cleanup_not_completed, descriptor_drain_not_ended, descriptor_lease_not_renewed, + entry_kind, fts_index_update_failed, held_entry_debt_not_recorded, + history_compaction_not_applied, ilp_invalid_utf8_drop, ilp_line_read_drop, + index_rebuild_not_installed, metadata_apply_wedged, orphaned_index_entry_after_delete, + quota_row_invalid, quota_row_undecodable, quota_row_write_failed, quota_scope_purge_incomplete, + quota_scope_replay_aborted, raft_entries_reapplied, raft_entry_reapplied, + redo_abandon_given_up, redo_entry_malformed, redo_snapshot_debt_not_recorded, + redo_stream_lost_here, replay_record_unapplied, retention_autowire_orphaned, + scope_quota_not_installed, strict_row_image_unrendered, strict_row_undecodable, + synonym_group_not_applied, timeseries_flush_not_installed, timeseries_idle_flush_failed, + timeseries_partition_unreadable, timeseries_row_image_undecodable, vector_build_failed, + vector_builder_disconnected, vector_builder_spawn_failed, vector_index_not_applied, + vector_payload_undecodable, vector_rebuild_unreadable, vshard_departure_purge_failed, + wal_archival_failed_truncation_held, write_acked_without_durability, write_window_held, + write_window_leaked, }; diff --git a/nodedb/src/diag/recording/lease.rs b/nodedb/src/diag/recording/lease.rs index 867de1d9e..7e9019016 100644 --- a/nodedb/src/diag/recording/lease.rs +++ b/nodedb/src/diag/recording/lease.rs @@ -1,6 +1,7 @@ // SPDX-License-Identifier: BUSL-1.1 -//! Capture sites for descriptor leases the renewal loop could not maintain. +//! Capture sites for descriptor leases, drains and DDL preparation leases +//! that were not maintained or cleaned up. use faultbox::{Capture, EventKind, error_chain_of}; @@ -35,3 +36,49 @@ pub fn descriptor_lease_not_renewed( .with_backtrace() .emit(); } + +/// Report a DDL preparation lease whose failure cleanup (cancel its pending +/// record, release it) did not apply on any attempt. Called only from the +/// background lease releaser's give-up arm. +pub fn ddl_cleanup_not_completed(err: &crate::Error, step: &'static str, token: u64) { + let class = error_class(err); + let ctx = context::DdlCleanupNotCompleted { + step, + token, + error_class: &class, + }; + let _ = Capture::new( + EventKind::Error, + "DDL cleanup: a failed DDL's lease was neither cancelled nor released", + ) + .error_chain(error_chain_of(err)) + .domain(&ctx) + .with_backtrace() + .emit(); +} + +/// Report a descriptor drain whose holder gave up and whose end did not apply +/// on any attempt. Called only from the background lease releaser's give-up +/// arm. +pub fn descriptor_drain_not_ended( + err: &crate::Error, + descriptor: &nodedb_cluster::DescriptorId, + owner: &nodedb_cluster::DrainOwner, +) { + let class = error_class(err); + let descriptor = format!("{descriptor:?}"); + let owner = format!("{owner:?}"); + let ctx = context::DescriptorDrainNotEnded { + descriptor: &descriptor, + owner: &owner, + error_class: &class, + }; + let _ = Capture::new( + EventKind::Error, + "descriptor drain: an abandoned drain did not end", + ) + .error_chain(error_chain_of(err)) + .domain(&ctx) + .with_backtrace() + .emit(); +} diff --git a/nodedb/src/diag/recording/mod.rs b/nodedb/src/diag/recording/mod.rs index a67be43a1..c60c5edf8 100644 --- a/nodedb/src/diag/recording/mod.rs +++ b/nodedb/src/diag/recording/mod.rs @@ -59,7 +59,9 @@ pub use event_image::{strict_row_image_unrendered, timeseries_row_image_undecoda pub use held_entry::{apply_panicked, held_entry_debt_not_recorded, redo_entry_malformed}; pub use index_rebuild::{IndexRebuildTarget, index_rebuild_not_installed}; pub use ingest::{ilp_invalid_utf8_drop, ilp_line_read_drop}; -pub use lease::descriptor_lease_not_renewed; +pub use lease::{ + ddl_cleanup_not_completed, descriptor_drain_not_ended, descriptor_lease_not_renewed, +}; pub use outcome_floor::{write_window_held, write_window_leaked}; pub use quota::{ quota_row_invalid, quota_row_undecodable, quota_row_write_failed, quota_scope_purge_incomplete, diff --git a/nodedb/src/main.rs b/nodedb/src/main.rs index 4251004ad..b3ee93f4a 100644 --- a/nodedb/src/main.rs +++ b/nodedb/src/main.rs @@ -158,6 +158,7 @@ async fn server_main() -> anyhow::Result<()> { raft_gate, schema_gate, sanity_gate, + data_plane_replay_gate, data_groups_gate, transport_gate, warm_peers_gate, @@ -282,6 +283,7 @@ async fn server_main() -> anyhow::Result<()> { raft_ready_rx, replay_done, nodedb::bootstrap::cluster_ready::ClusterReadyGates { + data_plane_replay_gate, raft_gate, schema_gate, sanity_gate, diff --git a/nodedb/src/main_boot/gates.rs b/nodedb/src/main_boot/gates.rs index c5820cb86..45b8cf881 100644 --- a/nodedb/src/main_boot/gates.rs +++ b/nodedb/src/main_boot/gates.rs @@ -14,6 +14,7 @@ pub(crate) struct StartupGates { pub(crate) raft_gate: ReadyGate, pub(crate) schema_gate: ReadyGate, pub(crate) sanity_gate: ReadyGate, + pub(crate) data_plane_replay_gate: ReadyGate, pub(crate) data_groups_gate: ReadyGate, pub(crate) transport_gate: ReadyGate, pub(crate) warm_peers_gate: ReadyGate, @@ -29,6 +30,8 @@ pub(crate) fn register_startup_gates(startup_seq: &StartupSequencer) -> StartupG let wal_gate = startup_seq.register_gate(StartupPhase::WalRecovery, "wal"); let catalog_gate = startup_seq.register_gate(StartupPhase::ClusterCatalogOpen, "cluster-catalog"); + let data_plane_replay_gate = + startup_seq.register_gate(StartupPhase::DataPlaneReplay, "data-plane-replay"); let raft_gate = startup_seq.register_gate(StartupPhase::RaftMetadataReplay, "raft-metadata-replay"); let schema_gate = @@ -50,6 +53,7 @@ pub(crate) fn register_startup_gates(startup_seq: &StartupSequencer) -> StartupG raft_gate, schema_gate, sanity_gate, + data_plane_replay_gate, data_groups_gate, transport_gate, warm_peers_gate, diff --git a/nodedb/tests/crash_boot_slow_replay.rs b/nodedb/tests/crash_boot_slow_replay.rs new file mode 100644 index 000000000..7e150d347 --- /dev/null +++ b/nodedb/tests/crash_boot_slow_replay.rs @@ -0,0 +1,193 @@ +// SPDX-License-Identifier: BUSL-1.1 + +//! A restart whose WAL replay outlasts the statement deadline still boots. +//! +//! A Data Plane core drains no request until its WAL replay ends, while every +//! request's deadline runs from its dispatch. Boot work that dispatched to the +//! cores during replay timed out against it: the schema rehydrate failed the +//! boot, and a committed Calvin redo reported as not applied halted the +//! scheduler for good. Nothing was checkpointed, so every later boot replayed +//! the same tail and failed the same way. +//! +//! Each test lowers the statement deadline, slows replay past it with the +//! `replay::before_engine_passes` fail point, and checks the restart boots +//! with every row present exactly once. Requires `--features failpoints`. + +#![cfg(feature = "failpoints")] + +mod crash_harness; + +use std::time::{Duration, Instant}; + +use crash_harness::{CrashHarness, diagnostics}; + +/// The statement deadline the server boots with. +const DEADLINE_SECS: u64 = 5; + +/// How long the restart holds each core's replay: past the deadline. +const SLOW_REPLAY_MS: u64 = 8_000; + +/// The smallest redo entry limit the config takes, so a commit of a few +/// hundred kilobytes travels as a chunked redo stream. +const ENTRY_LIMIT: usize = 64 * 1024; + +/// Rows one statement inserts, and statements the chunked commit runs. +const ROWS_PER_STATEMENT: usize = 100; +const STATEMENTS: usize = 6; + +/// Bounded wait for the injected abort. +const CRASH_TIMEOUT: Duration = Duration::from_secs(60); + +fn harness() -> CrashHarness { + let h = CrashHarness::new(); + let config = h.data_dir().join("slow_replay.toml"); + std::fs::write( + &config, + format!( + "[tuning.network]\ndefault_deadline_secs = {DEADLINE_SECS}\n\n\ + [tuning.calvin]\nmax_redo_entry_bytes = {ENTRY_LIMIT}\n" + ), + ) + .expect("write config file"); + let path = config.to_string_lossy().to_string(); + h.with_env("NODEDB_CONFIG", &path) +} + +/// Restart with every core's replay held past the statement deadline, and +/// wait for ready. +fn reopen_with_slow_replay(h: &mut CrashHarness) { + h.set_env( + "NODEDB_FAILPOINTS", + &format!("replay::before_engine_passes=sleep({SLOW_REPLAY_MS})"), + ); + let started = Instant::now(); + h.spawn(); + h.wait_ready_extended(); + // The sleep fail point logs nothing, so the boot's length shows it held + // the replay. A shorter boot proves nothing about a slow replay. + assert!( + started.elapsed() >= Duration::from_millis(SLOW_REPLAY_MS), + "the restart was ready in {:?}, under the {SLOW_REPLAY_MS}ms replay hold, so the \ + slowed replay never ran", + started.elapsed() + ); +} + +async fn count(h: &CrashHarness, coll: &str) -> usize { + let rows = h + .query_col_idx(&format!("SELECT COUNT(*) FROM {coll}"), 0) + .await; + assert_eq!(rows.len(), 1, "expected one COUNT(*) row, got {rows:?}"); + rows[0].parse().expect("COUNT(*) is a number") +} + +fn assert_no_calvin_halt(h: &CrashHarness) { + let log = h.server_log(); + assert!( + !log.contains("apply halted"), + "a Calvin scheduler halted during the slow-replay boot.{}\n{}", + h.keep_data_dir_note(), + diagnostics::log_tail_section(&log) + ); +} + +/// The schema rehydrate dispatches to every core. It waits for the replay +/// instead of timing out against it. +#[tokio::test(flavor = "multi_thread")] +async fn a_strict_collection_boots_through_a_slow_replay() { + let mut h = harness(); + h.spawn(); + h.wait_ready_extended(); + h.exec( + "CREATE COLLECTION slow_strict (id STRING PRIMARY KEY, n INT) \ + WITH (engine='document_strict')", + ) + .await; + for id in 0..20 { + h.exec(&format!( + "INSERT INTO slow_strict (id, n) VALUES ('r{id}', {id})" + )) + .await; + } + h.kill_9(); + + reopen_with_slow_replay(&mut h); + + assert_eq!(count(&h, "slow_strict").await, 20); + let typed = h + .query_col_idx("SELECT n FROM slow_strict WHERE id = 'r7'", 0) + .await; + assert_eq!(typed, vec!["7".to_string()]); + assert_no_calvin_halt(&h); +} + +/// Run one large transaction on its own connection. Returns whether COMMIT +/// answered success: a server that aborts mid-commit answers nothing. +async fn commit_large_transaction(h: &CrashHarness, coll: &str) -> bool { + let (client, connection) = tokio_postgres::connect(&h.pgwire_conn_str(), tokio_postgres::NoTls) + .await + .expect("connect"); + let conn = tokio::spawn(async move { + let _ = connection.await; + }); + let filler = "x".repeat(400); + client.simple_query("BEGIN").await.expect("BEGIN"); + for statement in 0..STATEMENTS { + let values: Vec = (0..ROWS_PER_STATEMENT) + .map(|row| { + let id = statement * ROWS_PER_STATEMENT + row; + format!("('c{id}', '{filler}')") + }) + .collect(); + client + .simple_query(&format!( + "INSERT INTO {coll} (id, value) VALUES {}", + values.join(", ") + )) + .await + .expect("INSERT inside the transaction"); + } + let committed = client.simple_query("COMMIT").await.is_ok(); + drop(client); + let _ = conn.await; + committed +} + +/// A committed redo re-delivered at boot installs once the cores replayed. It +/// never reports as not applied, so no scheduler halts. +#[tokio::test(flavor = "multi_thread")] +async fn a_committed_redo_installs_through_a_slow_replay() { + let fail_point = "redo_chunk::before_final_install"; + let mut h = harness(); + h.set_env("NODEDB_FAILPOINTS", &format!("{fail_point}=abort")); + h.spawn(); + h.wait_ready_extended(); + h.exec( + "CREATE COLLECTION slow_redo (id STRING PRIMARY KEY, value STRING) \ + WITH (engine='document_strict')", + ) + .await; + let committed = commit_large_transaction(&h, "slow_redo").await; + assert!( + !committed, + "COMMIT answered success, but the server was armed to abort at `{fail_point}`" + ); + h.await_self_crash(CRASH_TIMEOUT); + let log = h.server_log(); + assert!( + log.contains(&format!("fail_point aborting process: {fail_point}")), + "the server exited, but not at `{fail_point}`, so this test proves nothing about a \ + re-delivered redo.{}\n{}", + h.keep_data_dir_note(), + diagnostics::log_tail_section(&log) + ); + + reopen_with_slow_replay(&mut h); + + assert_eq!( + count(&h, "slow_redo").await, + ROWS_PER_STATEMENT * STATEMENTS, + "the re-delivered redo must install every row exactly once" + ); + assert_no_calvin_halt(&h); +} diff --git a/nodedb/tests/crash_dropped_collection_reclaim.rs b/nodedb/tests/crash_dropped_collection_reclaim.rs index afe515465..6c3fe5d26 100644 --- a/nodedb/tests/crash_dropped_collection_reclaim.rs +++ b/nodedb/tests/crash_dropped_collection_reclaim.rs @@ -16,6 +16,9 @@ use crash_harness::CrashHarness; const CHECKPOINT_WAIT: Duration = Duration::from_secs(20); +/// Long enough for several 1-second checkpoint cycles after a restart. +const POST_RESTART_CHECKPOINT_WAIT: Duration = Duration::from_secs(3); + /// Server env for a harness that must actually produce a checkpoint file /// within the test's runtime. A 1-second interval guarantees at least one /// flush fires well inside the `CHECKPOINT_WAIT` poll budget below. @@ -162,6 +165,8 @@ async fn dropped_vector_collection_does_not_resurrect_after_restart() { h.reopen(); assert_collection_missing(&h, &format!("SELECT id FROM {COLLECTION}"), COLLECTION).await; + // Same rationale as the spatial test below. + tokio::time::sleep(POST_RESTART_CHECKPOINT_WAIT).await; let leaked = any_file_name_contains(&ckpt_root, &needle); assert!( @@ -215,6 +220,10 @@ async fn dropped_spatial_collection_does_not_resurrect_after_restart() { h.reopen(); assert_collection_missing(&h, &format!("SELECT id FROM {COLLECTION}"), COLLECTION).await; + // Let the 1-second checkpoint cycles run after the restart. A restarted + // core that still holds the dropped collection's index writes its files + // again on the first cycle, so a check before any cycle proves nothing. + tokio::time::sleep(POST_RESTART_CHECKPOINT_WAIT).await; let leaked = any_file_name_contains(&ckpt_root, COLLECTION); assert!( @@ -225,3 +234,75 @@ async fn dropped_spatial_collection_does_not_resurrect_after_restart() { walk_files(&ckpt_root) ); } + +/// The columnar checkpoint filename's collection part: the name's bytes as +/// lowercase hex, as the write path encodes it. +fn hex_name(collection: &str) -> String { + collection.bytes().map(|b| format!("{b:02x}")).collect() +} + +/// Same claim, plain columnar engine. The columnar checkpoint is what every +/// columnar-family collection restores from, `spatial` included, so a file +/// left behind resurrects the dropped rows and a same-name CREATE inherits +/// them. +#[tokio::test(flavor = "multi_thread")] +async fn dropped_columnar_collection_does_not_resurrect_after_restart() { + const COLLECTION: &str = "crashdropcolumnar"; + + let mut h = checkpoint_every_second(); + h.spawn(); + h.wait_ready(); + + h.exec(&format!( + "CREATE COLLECTION {COLLECTION} (id TEXT, n INT) WITH (engine='columnar')" + )) + .await; + for id in 0..3 { + h.exec(&format!( + "INSERT INTO {COLLECTION} (id, n) VALUES ('r{id}', {id})" + )) + .await; + } + + let ckpt_root = h.data_dir().join("columnar-ckpt"); + let needle = format!("-coll-{}.ckpt", hex_name(COLLECTION)); + let deadline = Instant::now() + CHECKPOINT_WAIT; + while any_file_name_contains(&ckpt_root, &needle).is_none() { + assert!( + Instant::now() < deadline, + "no columnar checkpoint file named '{COLLECTION}' within {CHECKPOINT_WAIT:?}, so the \ + test can prove nothing about reclaim (files: {:?})", + walk_files(&ckpt_root) + ); + tokio::time::sleep(Duration::from_millis(100)).await; + } + + h.exec(&format!("DROP COLLECTION {COLLECTION} PURGE")).await; + h.kill_9(); + h.reopen(); + + assert_collection_missing(&h, &format!("SELECT id FROM {COLLECTION}"), COLLECTION).await; + // Same rationale as the spatial test above. + tokio::time::sleep(POST_RESTART_CHECKPOINT_WAIT).await; + let leaked = any_file_name_contains(&ckpt_root, &needle); + assert!( + leaked.is_none(), + "columnar checkpoint file for dropped collection '{COLLECTION}' is still on disk after \ + DROP COLLECTION + restart: {leaked:?} (all files under {}: {:?})", + ckpt_root.display(), + walk_files(&ckpt_root) + ); + + h.exec(&format!( + "CREATE COLLECTION {COLLECTION} (id TEXT, n INT) WITH (engine='columnar')" + )) + .await; + let count = h + .query_col_idx(&format!("SELECT COUNT(*) FROM {COLLECTION}"), 0) + .await; + assert_eq!( + count, + vec!["0".to_string()], + "a same-name CREATE after the drop inherited the dropped collection's rows" + ); +} diff --git a/nodedb/tests/crash_named_database_catalog_refs.rs b/nodedb/tests/crash_named_database_catalog_refs.rs new file mode 100644 index 000000000..ffeff56cd --- /dev/null +++ b/nodedb/tests/crash_named_database_catalog_refs.rs @@ -0,0 +1,73 @@ +// SPDX-License-Identifier: BUSL-1.1 + +//! Catalog objects that reference a collection in a named database must not +//! fail the boot sanity check. +//! +//! An RLS policy stores its collection database-qualified +//! (`"{database_id}/{name}"`), while the catalog keys the collection by its +//! bare name and database. The boot integrity check must resolve the policy's +//! collection in the policy's own database, or it reports a dangling reference +//! and every boot refuses to start. A change stream on the same collection +//! rides along, since its check is scoped the same way. + +mod crash_harness; + +use crash_harness::CrashHarness; + +const DATABASE: &str = "named_refs_db"; +const COLLECTION: &str = "named_refs_pat"; + +#[tokio::test(flavor = "multi_thread")] +async fn a_named_database_rls_policy_and_change_stream_survive_restart() { + let mut h = CrashHarness::new(); + h.spawn(); + h.wait_ready(); + + h.exec(&format!("CREATE DATABASE {DATABASE}")).await; + h.exec_in( + DATABASE, + &format!("CREATE COLLECTION {COLLECTION} (id TEXT PRIMARY KEY, owner TEXT)"), + ) + .await; + h.exec_in( + DATABASE, + &format!("INSERT INTO {COLLECTION} (id, owner) VALUES ('r1', 'alice')"), + ) + .await; + h.exec_in( + DATABASE, + &format!( + "CREATE RLS POLICY named_refs_owner ON {COLLECTION} FOR READ \ + USING (owner = 'alice')" + ), + ) + .await; + h.exec_in( + DATABASE, + &format!("CREATE CHANGE STREAM named_refs_stream ON {COLLECTION}"), + ) + .await; + + h.kill_9(); + // `reopen` panics with the server output when the boot fails, which is + // how the dangling-reference sanity check showed up. + h.reopen(); + + let policies = h.query_col_idx_in(DATABASE, "SHOW RLS POLICIES", 0).await; + assert!( + policies.iter().any(|name| name == "named_refs_owner"), + "the RLS policy is missing after the restart: {policies:?}" + ); + let ids = h + .query_col_idx_in(DATABASE, &format!("SELECT id FROM {COLLECTION}"), 0) + .await; + assert_eq!( + ids, + vec!["r1".to_string()], + "the row is missing after the restart" + ); + + // A second restart boots from the state the first restart left behind. + h.kill_9(); + h.reopen(); +} diff --git a/nodedb/tests/crash_replay_fail_stop.rs b/nodedb/tests/crash_replay_fail_stop.rs index 25f63b8d3..a0249cff0 100644 --- a/nodedb/tests/crash_replay_fail_stop.rs +++ b/nodedb/tests/crash_replay_fail_stop.rs @@ -16,9 +16,11 @@ //! //! This test forces exactly that: a WAL-replay fail point panics the sole //! Data Plane core's thread before it signals replay completion, which drops -//! the `replay_done` sender. `bootstrap/cluster_ready.rs` turns that into -//! "data plane core exited before signalling WAL replay completion" and fails -//! the `data-groups-replay` gate, which must fail-stop the whole process. +//! the `replay_done` sender. `bootstrap/cluster_ready.rs` waits for every +//! core's replay before any boot step that dispatches to a core, turns the +//! dropped sender into "data plane core exited before signalling WAL replay +//! completion", and fails the `data-plane-replay` gate, which must fail-stop +//! the whole process. //! Requires `--features failpoints`. #![cfg(feature = "failpoints")] @@ -28,15 +30,10 @@ mod crash_harness; use crash_harness::CrashHarness; use std::time::Duration; -/// Bound on how long a fail-stopped boot is given to actually exit. -/// -/// A dead core is not detected directly: the Control Plane notices only when a -/// dispatch to it stops being answered, and the schema-register barrier that -/// runs first bounds that wait at `default_deadline_secs` (30s). The exit -/// therefore lands a few seconds the far side of that barrier, so anything at -/// or below 30s measures the barrier rather than the fail-stop and reports a -/// failure the code does not have. -const BOOT_FAILURE_TIMEOUT: Duration = Duration::from_secs(90); +/// Bound on how long a fail-stopped boot is given to actually exit. The +/// replay wait sees the dropped sender at once, so this bounds only the boot +/// up to that wait and the exit after it. +const BOOT_FAILURE_TIMEOUT: Duration = Duration::from_secs(60); /// A readiness-gate failure during WAL replay must fail-stop the process /// rather than leave it alive with every listener port closed. @@ -88,15 +85,11 @@ async fn replay_readiness_gate_failure_fails_the_boot() { "the server panicked and exited, but without the fail-stop diagnosis — an operator \ would see a dead process and no stated reason.\nServer output:\n{log}" ); - // The dead core is reported by the schema-register barrier, not by the - // replay-done signal: `rehydrate_schema_registry` dispatches to the Data - // Plane before `await_cluster_ready` reaches its replay wait. The barrier - // fails fast — the dispatcher's dead-core sweep synthesises an error for - // every outstanding request rather than letting the deadline expire. + // The dead core is reported by the replay wait: no boot step dispatches + // to a core before every core signalled its replay. assert!( - log.contains("core-0 is gone; the request can never be executed"), - "the server fail-stopped, but not through the dead-core path this test targets — \ - the schema-register barrier's dead-core error is missing.\n\ - Server output:\n{log}" + log.contains("data plane core exited before signalling WAL replay completion"), + "the server fail-stopped, but not through the replay wait this test targets — its \ + dead-core error is missing.\nServer output:\n{log}" ); } diff --git a/nodedb/tests/wire/cases/mod.rs b/nodedb/tests/wire/cases/mod.rs index 44b3c3cce..3faa8610d 100644 --- a/nodedb/tests/wire/cases/mod.rs +++ b/nodedb/tests/wire/cases/mod.rs @@ -461,6 +461,7 @@ mod trigger_statement_txn; mod truncate_engine_conformance; mod truncate_engine_conformance_columnar_family; mod txn_ddl_commit_registry_sync; +mod txn_guarded_update_with_other_write; mod unbounded_reads; mod user_transaction; mod vector_dimension_errors; diff --git a/nodedb/tests/wire/cases/sql_sequences.rs b/nodedb/tests/wire/cases/sql_sequences.rs index 725bcf9bb..7c56a8775 100644 --- a/nodedb/tests/wire/cases/sql_sequences.rs +++ b/nodedb/tests/wire/cases/sql_sequences.rs @@ -375,3 +375,126 @@ async fn duplicate_constant_column_names_keep_their_own_cells() { display name collapses them to the last value" ); } + +/// The same repeated output name through the extended protocol +/// (Parse/Bind/Execute). Describe announces the columns, and the projection +/// built from it must keep the planner's per-column lookup keys, or both +/// columns read the first cell. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn duplicate_constant_column_names_keep_their_own_cells_extended_protocol() { + let server = TestServer::start().await; + server + .exec("CREATE SEQUENCE dup_cell_ext_seq START 1 INCREMENT 1") + .await + .unwrap(); + + let rows = server + .client + .query( + "SELECT nextval('dup_cell_ext_seq'), nextval('dup_cell_ext_seq')", + &[], + ) + .await + .expect("a prepared from-less projection repeating an output name must run"); + assert_eq!(rows.len(), 1, "expected exactly one constant row"); + let cells: (i64, i64) = (rows[0].get(0), rows[0].get(1)); + assert_eq!( + cells, + (1, 2), + "each duplicate column must keep its own cell over the extended protocol" + ); +} + +/// A join that projects two columns of one bare name, and a projection that +/// repeats an aggregate, keep their own cells over the extended protocol. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn duplicate_join_and_aggregate_names_keep_their_own_cells_extended_protocol() { + let server = TestServer::start().await; + for sql in [ + "CREATE COLLECTION dup_w (id TEXT PRIMARY KEY, b_id TEXT) WITH (engine='document_strict')", + "CREATE COLLECTION dup_b (id TEXT PRIMARY KEY, n INT) WITH (engine='document_strict')", + "INSERT INTO dup_b (id, n) VALUES ('b1', 5)", + "INSERT INTO dup_b (id, n) VALUES ('b2', 7)", + "INSERT INTO dup_w (id, b_id) VALUES ('w1', 'b1')", + ] { + server.exec(sql).await.unwrap(); + } + + let rows = server + .client + .query( + "SELECT dup_w.id, dup_b.id FROM dup_w JOIN dup_b ON dup_w.b_id = dup_b.id", + &[], + ) + .await + .expect("a prepared join repeating a bare column name must run"); + assert_eq!(rows.len(), 1, "expected one joined row"); + let ids: (String, String) = (rows[0].get(0), rows[0].get(1)); + assert_eq!( + ids, + ("w1".to_string(), "b1".to_string()), + "each side of the join keeps its own id cell" + ); + + let rows = server + .client + .query("SELECT count(*), sum(n), count(*) FROM dup_b", &[]) + .await + .expect("a prepared projection repeating an aggregate must run"); + assert_eq!(rows.len(), 1, "expected one aggregate row"); + let counts: (i64, i64) = (rows[0].get(0), rows[0].get(2)); + assert_eq!(counts, (2, 2), "both count(*) columns read the count"); +} + +/// Describing a statement plans it without executing it: Parse announces a +/// `nextval` column's type and consumes no value, so the first execution +/// still gets the sequence's start. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn describing_nextval_consumes_no_value() { + let server = TestServer::start().await; + server + .exec("CREATE SEQUENCE describe_only_seq START 1 INCREMENT 1") + .await + .unwrap(); + + for _ in 0..2 { + let statement = server + .client + .prepare("SELECT nextval('describe_only_seq')") + .await + .expect("prepare a nextval projection"); + assert_eq!( + statement.columns().len(), + 1, + "Describe announces the nextval column" + ); + assert_eq!( + statement.columns()[0].type_(), + &tokio_postgres::types::Type::INT8 + ); + } + + let rows = server + .query_rows("SELECT nextval('describe_only_seq')") + .await + .unwrap(); + assert_eq!( + rows, + vec![vec!["1".to_string()]], + "describing the statement consumed a sequence value" + ); +} + +/// Describing a `nextval` of a sequence that does not exist reports the +/// missing sequence, as executing it does. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn describing_nextval_of_a_missing_sequence_is_refused() { + let server = TestServer::start().await; + let refused = server + .client + .prepare("SELECT nextval('no_such_describe_seq')") + .await; + let error = refused.expect_err("Parse of a nextval on a missing sequence must be refused"); + let code = error.code().map(|c| c.code().to_string()); + assert_eq!(code.as_deref(), Some("42704"), "{error}"); +} diff --git a/nodedb/tests/wire/cases/txn_guarded_update_with_other_write.rs b/nodedb/tests/wire/cases/txn_guarded_update_with_other_write.rs new file mode 100644 index 000000000..d60cd0e25 --- /dev/null +++ b/nodedb/tests/wire/cases/txn_guarded_update_with_other_write.rs @@ -0,0 +1,103 @@ +// SPDX-License-Identifier: BUSL-1.1 + +//! A guarded UPDATE together with another write in one transaction commits. +//! +//! `WHERE id = 'b1' AND state = 'pending'` names no primary-key point, so the +//! UPDATE plans as a predicate write. With an INSERT on another collection the +//! commit spans vShards and goes through Calvin. The staged predicate write +//! must carry its predicted rows into the Calvin stage, or every COMMIT aborts +//! with a retry hint that no retry satisfies. Both protocols run the exact +//! statements from the report. + +use crate::harness::TestServer; + +async fn setup(server: &TestServer, bk: &str, le: &str) { + server + .exec(&format!( + "CREATE COLLECTION {bk} (id TEXT PRIMARY KEY, state TEXT) \ + WITH (engine='document_strict')" + )) + .await + .unwrap(); + server + .exec(&format!( + "CREATE COLLECTION {le} (id TEXT PRIMARY KEY, amount_sen BIGINT) \ + WITH (engine='document_strict')" + )) + .await + .unwrap(); + server + .exec(&format!( + "INSERT INTO {bk} (id, state) VALUES ('b1', 'pending')" + )) + .await + .unwrap(); +} + +async fn assert_committed(server: &TestServer, bk: &str, le: &str) { + let state = server + .query_text(&format!("SELECT state FROM {bk} WHERE id = 'b1'")) + .await + .unwrap(); + assert_eq!( + state, + vec!["confirmed".to_string()], + "the guarded UPDATE applied" + ); + let amount = server + .query_text(&format!("SELECT amount_sen FROM {le} WHERE id = 'e1'")) + .await + .unwrap(); + assert_eq!(amount, vec!["1".to_string()], "the INSERT applied"); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn a_guarded_update_with_another_write_commits_simple_protocol() { + let server = TestServer::start().await; + setup(&server, "guard_bk_s", "guard_le_s").await; + + server.exec("BEGIN").await.unwrap(); + server + .exec("INSERT INTO guard_le_s (id, amount_sen) VALUES ('e1', 1)") + .await + .unwrap(); + server + .exec("UPDATE guard_bk_s SET state = 'confirmed' WHERE id = 'b1' AND state = 'pending'") + .await + .unwrap(); + server + .exec("COMMIT") + .await + .unwrap_or_else(|e| panic!("COMMIT of a guarded UPDATE plus an INSERT failed: {e}")); + + assert_committed(&server, "guard_bk_s", "guard_le_s").await; +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn a_guarded_update_with_another_write_commits_extended_protocol() { + let server = TestServer::start().await; + setup(&server, "guard_bk_x", "guard_le_x").await; + + let client = &server.client; + client.execute("BEGIN", &[]).await.unwrap(); + client + .execute( + "INSERT INTO guard_le_x (id, amount_sen) VALUES ($1, $2)", + &[&"e1", &1_i64], + ) + .await + .unwrap(); + client + .execute( + "UPDATE guard_bk_x SET state = 'confirmed' WHERE id = $1 AND state = $2", + &[&"b1", &"pending"], + ) + .await + .unwrap(); + client + .execute("COMMIT", &[]) + .await + .unwrap_or_else(|e| panic!("COMMIT of a guarded UPDATE plus an INSERT failed: {e}")); + + assert_committed(&server, "guard_bk_x", "guard_le_x").await; +}