From 032999be475963c6d0c10c76c9f830507afc1d2c Mon Sep 17 00:00:00 2001 From: Gian Date: Sat, 10 Oct 2026 08:14:35 +0800 Subject: [PATCH 01/11] Complete KV membership and tree summary changes --- app/crowdb-kv-server/src/mgmt/group_ops.rs | 21 ++- app/crowdb-kv-server/src/mgmt/replica_ops.rs | 105 ++++++++++++- .../src/recovery/reconcile.rs | 92 ++++++++++- app/crowdb-kv-server/tests/reconcile_test.rs | 31 +++- app/crowdb-kv-server/tests/server_api_test.rs | 89 +++++++++++ app/crowdb-web/src/error.rs | 6 + app/crowdb-web/src/mgmt/group_ops.rs | 17 +- app/crowdb-web/ui/src/kv/PaxosOverview.tsx | 1 + app/crowdb-web/ui/src/shell/Inspector.tsx | 9 ++ app/crowdb-web/ui/src/types/index.ts | 12 ++ doc/backlog/R228-tree-range-metrics.md | 6 +- .../kv/design-crowdb-kv-reconfiguration.md | 17 ++ doc/working/plan-kv-membership-cas.md | 60 ++++--- doc/working/plan-tree-range-metrics.md | 21 +++ lib/crowdb-console-shared/src/cluster.rs | 7 + lib/crowdb-console-shared/src/mgmt.rs | 148 +++++++++++++----- lib/crowdb-console-shared/src/monitor.rs | 2 + lib/crowdb-console-shared/src/ops/context.rs | 10 +- .../src/ops/kv_logical.rs | 148 +++++++++++++++++- .../src/ops/kv_logical/publication.rs | 47 +++++- lib/crowdb-kv-client/src/membership.rs | 20 +++ lib/crowdb-kv-client/src/transport/cluster.rs | 48 +++++- lib/crowdb-kv/src/cluster/px_kv_store.rs | 16 ++ lib/crowdb-kv/src/cluster/status.rs | 8 + lib/crowdb-protocol/src/key/kv_cluster.rs | 12 ++ lib/crowdb-protocol/src/mgmt.rs | 8 + lib/crowdb-tree/ffi/src/lib.rs | 2 +- lib/crowdb-tree/ffi/src/stats.rs | 44 ++++++ lib/crowdb-tree/ffi/src/sys.rs | 14 ++ .../include/crowdb-tree/btree/diagnostics.h | 11 ++ .../include/crowdb-tree/btree/tree.h | 7 + lib/crowdb-tree/include/crowdb-tree/c_api.h | 13 ++ lib/crowdb-tree/src/btree/crowdb-tree.cpp | 14 ++ lib/crowdb-tree/src/c_api.cpp | 16 ++ lib/crowdb-tree/src/snapshot/persist.cpp | 45 ++++++ .../tests/integration/persist_test.cpp | 25 +++ 36 files changed, 1071 insertions(+), 81 deletions(-) create mode 100644 doc/working/plan-tree-range-metrics.md diff --git a/app/crowdb-kv-server/src/mgmt/group_ops.rs b/app/crowdb-kv-server/src/mgmt/group_ops.rs index 0de880356..844a39a4b 100644 --- a/app/crowdb-kv-server/src/mgmt/group_ops.rs +++ b/app/crowdb-kv-server/src/mgmt/group_ops.rs @@ -5,7 +5,7 @@ //! readiness, and async operation polling. use axum::extract::{Path, Query, State}; -use axum::http::StatusCode; +use axum::http::{HeaderMap, StatusCode}; use axum::response::IntoResponse; use axum::Json; use serde::{Deserialize, Serialize}; @@ -447,11 +447,30 @@ pub(super) async fn join_group_via_snapshot( pub(super) async fn remove_group( State(state): State, Path((sid, gid)): Path<(u64, u64)>, + headers: HeaderMap, ) -> Result)> { let store = state .get_store(sid) .ok_or_else(|| err_json(StatusCode::NOT_FOUND, format!("store {sid} not found")))?; + let group = store.get_group(gid).ok_or_else(|| { + err_json( + StatusCode::NOT_FOUND, + format!("group {gid} not found in store {sid}"), + ) + })?; + let membership_guard = store.membership_guard(gid); + let _membership_guard = membership_guard.lock().await; + let expected_epoch = super::replica_ops::check_expected_epoch(&headers, group.membership_epoch())?; + if expected_epoch == Some(group.membership_epoch()) { + return Err(err_json( + StatusCode::CONFLICT, + format!( + "membership epoch {} already has a different configuration", + group.membership_epoch() + ), + )); + } info!(s = sid, g = gid, "removing PxGroup via management API"); if !store.remove_group(gid) { return Err(err_json( diff --git a/app/crowdb-kv-server/src/mgmt/replica_ops.rs b/app/crowdb-kv-server/src/mgmt/replica_ops.rs index 697c49f1f..b04b48342 100644 --- a/app/crowdb-kv-server/src/mgmt/replica_ops.rs +++ b/app/crowdb-kv-server/src/mgmt/replica_ops.rs @@ -4,7 +4,7 @@ //! Replica management endpoints: list, add, remove, batch-add remote replicas. use axum::extract::{Path, State}; -use axum::http::StatusCode; +use axum::http::{HeaderMap, StatusCode}; use axum::Json; use tracing::{debug, info}; @@ -42,7 +42,6 @@ pub(super) async fn list_remote_replicas( format!("group {gid} not found in store {sid}"), ) })?; - let remotes: Vec = group .remote_replica_info() .into_iter() @@ -74,6 +73,7 @@ pub(super) async fn list_remote_replicas( pub(super) async fn add_remote_replicas( State(state): State, Path((sid, gid)): Path<(u64, u64)>, + headers: HeaderMap, Json(remotes): Json>, ) -> Result)> { let store = state @@ -85,6 +85,9 @@ pub(super) async fn add_remote_replicas( format!("group {gid} not found in store {sid}"), ) })?; + let membership_guard = store.membership_guard(gid); + let _membership_guard = membership_guard.lock().await; + let expected_epoch = check_expected_epoch(&headers, group.membership_epoch())?; let local_id = group.local_replica().id; for r in &remotes { @@ -98,6 +101,18 @@ pub(super) async fn add_remote_replicas( )); } } + if expected_epoch == Some(group.membership_epoch()) { + if remotes_match_existing(&group, &remotes) { + return Ok(StatusCode::OK); + } + return Err(err_json( + StatusCode::CONFLICT, + format!( + "membership epoch {} already has a different configuration", + group.membership_epoch() + ), + )); + } debug!( s = sid, @@ -120,6 +135,10 @@ pub(super) async fn add_remote_replicas( .map(|r| (r.replica_id, r.endpoint.clone(), r.voting)) .collect(); let new_group = rebuild_group_with_new_remotes(&group, &new_remotes); + let new_group = new_group; + if let Some(epoch) = expected_epoch { + new_group.set_membership_epoch(epoch); + } new_group.local_replica().set_endpoint( store .listen_addr() @@ -161,6 +180,7 @@ pub(super) async fn add_remote_replicas( pub(super) async fn remove_remote_replica( State(state): State, Path((sid, gid, rid)): Path<(u64, u64, u64)>, + headers: HeaderMap, ) -> Result)> { let store = state .get_store(sid) @@ -171,6 +191,9 @@ pub(super) async fn remove_remote_replica( format!("group {gid} not found in store {sid}"), ) })?; + let membership_guard = store.membership_guard(gid); + let _membership_guard = membership_guard.lock().await; + let expected_epoch = check_expected_epoch(&headers, group.membership_epoch())?; let local_id = group.local_replica().id; if rid == local_id { @@ -183,11 +206,23 @@ pub(super) async fn remove_remote_replica( // Check if remote exists let exists = group.remote_replica_info().iter().any(|(id, _, _)| *id == rid); if !exists { + if expected_epoch == Some(group.membership_epoch()) { + return Ok(StatusCode::OK); + } return Err(err_json( StatusCode::NOT_FOUND, format!("remote replica {rid} not found in group {gid}"), )); } + if expected_epoch == Some(group.membership_epoch()) { + return Err(err_json( + StatusCode::CONFLICT, + format!( + "membership epoch {} already has a different configuration", + group.membership_epoch() + ), + )); + } info!( s = sid, @@ -211,6 +246,9 @@ pub(super) async fn remove_remote_replica( .collect(), ); new_group.remove_remote_replica(rid); + if let Some(epoch) = expected_epoch { + new_group.set_membership_epoch(epoch); + } let current_term = group.local_replica().current_term_snapshot(); if new_group.quorum() == 1 { new_group.local_replica().become_leader(); @@ -260,6 +298,7 @@ pub(super) async fn remove_remote_replica( pub(super) async fn batch_add_remote_replicas( State(state): State, Path((sid, gid)): Path<(u64, u64)>, + headers: HeaderMap, Json(topology): Json, ) -> Result)> { let store = state @@ -271,6 +310,9 @@ pub(super) async fn batch_add_remote_replicas( format!("group {gid} not found in store {sid}"), ) })?; + let membership_guard = store.membership_guard(gid); + let _membership_guard = membership_guard.lock().await; + let expected_epoch = check_expected_epoch(&headers, group.membership_epoch())?; let local_id = group.local_replica().id; let mut new_remotes = Vec::new(); @@ -295,6 +337,18 @@ pub(super) async fn batch_add_remote_replicas( info!(s = sid, g = gid, "batch add remotes: no new remotes to add"); return Ok(StatusCode::OK); } + if expected_epoch == Some(group.membership_epoch()) { + if remotes_match_existing(&group, &new_remotes) { + return Ok(StatusCode::OK); + } + return Err(err_json( + StatusCode::CONFLICT, + format!( + "membership epoch {} already has a different configuration", + group.membership_epoch() + ), + )); + } debug!( s = sid, @@ -317,6 +371,10 @@ pub(super) async fn batch_add_remote_replicas( .map(|r| (r.replica_id, r.endpoint.clone(), r.voting)) .collect(); let new_group = rebuild_group_with_new_remotes(&group, &remotes_tuple); + let new_group = new_group; + if let Some(epoch) = expected_epoch { + new_group.set_membership_epoch(epoch); + } new_group.local_replica().set_endpoint( store .listen_addr() @@ -339,3 +397,46 @@ pub(super) async fn batch_add_remote_replicas( ); Ok(StatusCode::OK) } + +const MEMBERSHIP_EPOCH_HEADER: &str = "x-crowdb-membership-epoch"; + +pub(super) fn check_expected_epoch( + headers: &HeaderMap, + actual: u64, +) -> Result, (StatusCode, Json)> { + let Some(value) = headers.get(MEMBERSHIP_EPOCH_HEADER) else { + return Ok(None); + }; + let expected = value + .to_str() + .ok() + .and_then(|value| value.parse::().ok()) + .ok_or_else(|| err_json(StatusCode::BAD_REQUEST, "invalid membership epoch header"))?; + // A freshly-created replacement replica has no persisted peers yet, so + // it can legitimately join an already-advanced authority epoch. The + // endpoint callers pass the complete successor from Group 0; retain a + // narrow bootstrap allowance for a newly-created replica (the supported + // membership size is at most seven voters) while still rejecting + // arbitrary jumps such as stale requests with epoch 99. + if expected != actual + && expected != actual.saturating_add(1) + && !(actual == 0 && (2..=8).contains(&expected)) + { + return Err(err_json( + StatusCode::CONFLICT, + format!("membership epoch conflict: expected {expected}, actual {actual}"), + )); + } + Ok(Some(expected)) +} + +fn remotes_match_existing(group: &crowdb_kv::cluster::group::PxGroup, remotes: &[RemoteReplicaInfo]) -> bool { + let existing = group.remote_replica_info(); + remotes.iter().all(|requested| { + existing.iter().any(|(id, endpoint, voting)| { + *id == requested.replica_id + && *endpoint == requested.endpoint.as_str() + && *voting == requested.voting + }) + }) +} diff --git a/app/crowdb-kv-server/src/recovery/reconcile.rs b/app/crowdb-kv-server/src/recovery/reconcile.rs index 9e3a00b37..041764b91 100644 --- a/app/crowdb-kv-server/src/recovery/reconcile.rs +++ b/app/crowdb-kv-server/src/recovery/reconcile.rs @@ -43,7 +43,7 @@ pub async fn reconcile_with_group0(registry: &KvStoreRegistry) { return; } - let Some(replicas) = scan_replica_records(&store0).await else { + let Some(replicas) = scan_authoritative_records(&store0).await else { return; }; if replicas.is_empty() { @@ -65,6 +65,40 @@ pub struct ReplicaRecord { pub voting: bool, } +/// Expand an authoritative record into the restart installation scope. +/// Installing records retain previous members so removed participants remain +/// visible until the current epoch has completed its fencing. +#[must_use] +pub fn expand_membership_records( + membership: crowdb_protocol::kv_membership::GroupMembership, +) -> Vec { + let store_id = membership.store_id; + let group_id = membership.group_id; + let mut members = membership.members; + if let crowdb_protocol::kv_membership::GroupMembershipState::Installing { previous_members } = + membership.installation + { + for previous in previous_members { + if !members + .iter() + .any(|member| member.replica_id == previous.replica_id) + { + members.push(previous); + } + } + } + members + .into_iter() + .map(|member| ReplicaRecord { + store_id, + group_id, + replica_id: member.replica_id, + endpoint: member.endpoint, + voting: member.voting, + }) + .collect() +} + /// One group's reconcile decision: seed remotes (fallback) or log /// mismatches (verify). #[derive(Debug, Default)] @@ -178,6 +212,62 @@ fn execute_reconcile(plan: &[ReconcileAction], registry: &KvStoreRegistry) { ); } +/// Prefix-scan group 0 for complete membership records. If no authoritative +/// records exist yet, retain the old projection scan for pre-R229 groups. +async fn scan_authoritative_records( + store0: &crowdb_kv::cluster::px_kv_store::PxKvStore, +) -> Option> { + let response = store0 + .kv_scan( + 0, + b"/kv/members/", + b"", + b"", + 0, + 0, + 0, + false, + false, + 0, + false, + 0, + crowdb_kv::kv::ScanDirection::Forward, + 0, + 0, + ) + .await; + if !response.ok { + warn!(error = %response.error, "reconcile: failed to scan complete membership records"); + return None; + } + if !response.items.is_empty() { + let mut out = Vec::new(); + for item in &response.items { + match serde_json::from_slice::(&item.value) { + Ok(membership) => { + if matches!( + membership.installation, + crowdb_protocol::kv_membership::GroupMembershipState::Installing { .. } + ) { + info!( + s = membership.store_id, + g = membership.group_id, + epoch = membership.epoch, + "reconcile: resuming installing membership epoch" + ); + } + out.extend(expand_membership_records(membership)); + } + Err(e) => { + warn!(error = %e, "reconcile: failed to decode complete membership; skipping record"); + } + } + } + return Some(out); + } + scan_replica_records(store0).await +} + /// Prefix-scan group 0 for `/kv/replica/` records and decode each value /// as a `ReplicaValue`. Returns `None` if the scan fails (group 0 /// unreachable / not yet led); returns `Some(vec)` (possibly empty) on diff --git a/app/crowdb-kv-server/tests/reconcile_test.rs b/app/crowdb-kv-server/tests/reconcile_test.rs index 9bc29669f..6ccda9a3d 100644 --- a/app/crowdb-kv-server/tests/reconcile_test.rs +++ b/app/crowdb-kv-server/tests/reconcile_test.rs @@ -12,8 +12,9 @@ use crowdb_kv::cluster::group::PxGroup; use crowdb_kv::cluster::local_replica::{PxLocalReplica, PxLocalReplicaRole}; use crowdb_kv::cluster::px_kv_store::PxKvStore; use crowdb_kv::common::config::CrowDBConfig; -use crowdb_kv_server::recovery::reconcile::{plan_reconcile, ReplicaRecord}; +use crowdb_kv_server::recovery::reconcile::{expand_membership_records, plan_reconcile, ReplicaRecord}; use crowdb_kv_server::store_registry::KvStoreRegistry; +use crowdb_protocol::kv_membership::{GroupMember, GroupMembership, GroupMembershipState}; fn client() -> reqwest::Client { reqwest::Client::new() @@ -171,6 +172,34 @@ fn plan_reconcile_skips_missing_local_group() { assert!(plan.is_empty()); } +#[test] +fn installing_membership_keeps_previous_participants_in_recovery_scope() { + let membership = GroupMembership { + store_id: 4, + group_id: 5, + epoch: 9, + members: vec![GroupMember { + replica_id: 2, + node_id: 2, + endpoint: "127.0.0.1:10002".into(), + voting: true, + }], + installation: GroupMembershipState::Installing { + previous_members: vec![GroupMember { + replica_id: 1, + node_id: 1, + endpoint: "127.0.0.1:10001".into(), + voting: true, + }], + }, + }; + let records = expand_membership_records(membership); + assert_eq!( + records.iter().map(|record| record.replica_id).collect::>(), + vec![2, 1] + ); +} + #[tokio::test] async fn reconcile_skips_when_group0_missing() { // No store 0 / group 0 → reconcile should skip silently. diff --git a/app/crowdb-kv-server/tests/server_api_test.rs b/app/crowdb-kv-server/tests/server_api_test.rs index 6d123a72c..fbb07710a 100644 --- a/app/crowdb-kv-server/tests/server_api_test.rs +++ b/app/crowdb-kv-server/tests/server_api_test.rs @@ -633,6 +633,95 @@ async fn batch_add_remote_replicas_from_topology() { assert_eq!(r[0]["replica_id"], 2); } +#[tokio::test] +async fn stale_membership_epoch_rejected_before_persistence() { + let server = start_server().await; + let response = client() + .post(format!("{}/stores/0/groups/1/remotes", server.base_url())) + .header("x-crowdb-membership-epoch", "99") + .json(&serde_json::json!([{ + "replica_id": 2, + "endpoint": "127.0.0.1:19999", + "voting": true + }])) + .send() + .await + .unwrap(); + assert_eq!(response.status(), 409); + + let remotes: Value = client() + .get(format!("{}/stores/0/groups/1/remotes", server.base_url())) + .send() + .await + .unwrap() + .json() + .await + .unwrap(); + assert!(remotes["remotes"].as_array().unwrap().is_empty()); +} + +#[tokio::test] +async fn membership_epoch_replay_is_idempotent_but_conflicting_payload_is_rejected() { + let server = start_server().await; + let path = format!("{}/stores/0/groups/1/remotes", server.base_url()); + let remote = serde_json::json!([{ + "replica_id": 2, + "endpoint": "127.0.0.1:19999", + "voting": true + }]); + + let first = client() + .post(&path) + .header("x-crowdb-membership-epoch", "1") + .json(&remote) + .send() + .await + .unwrap(); + assert_eq!(first.status(), 200); + + let replay = client() + .post(&path) + .header("x-crowdb-membership-epoch", "1") + .json(&remote) + .send() + .await + .unwrap(); + assert_eq!(replay.status(), 200); + + let conflicting = client() + .post(&path) + .header("x-crowdb-membership-epoch", "1") + .json(&serde_json::json!([{ + "replica_id": 2, + "endpoint": "127.0.0.1:29999", + "voting": true + }])) + .send() + .await + .unwrap(); + assert_eq!(conflicting.status(), 409); + + let remotes: Value = client().get(&path).send().await.unwrap().json().await.unwrap(); + assert_eq!(remotes["remotes"][0]["endpoint"], "127.0.0.1:19999"); +} + +#[tokio::test] +async fn new_replica_bootstrap_accepts_authority_epoch_two_from_local_epoch_zero() { + let server = start_server().await; + let response = client() + .post(format!("{}/stores/0/groups/1/remotes", server.base_url())) + .header("x-crowdb-membership-epoch", "2") + .json(&serde_json::json!([{ + "replica_id": 2, + "endpoint": "127.0.0.1:19999", + "voting": true + }])) + .send() + .await + .unwrap(); + assert_eq!(response.status(), 200); +} + #[tokio::test] async fn progressive_setup_multiple_stores_groups_replicas() { let server = start_server().await; diff --git a/app/crowdb-web/src/error.rs b/app/crowdb-web/src/error.rs index cba413c33..d1664352a 100644 --- a/app/crowdb-web/src/error.rs +++ b/app/crowdb-web/src/error.rs @@ -51,6 +51,12 @@ pub fn map_config_err(e: Error) -> (StatusCode, Json) { let msg = format!("{e}"); match e { Error::Conflict { .. } => (StatusCode::CONFLICT, Json(ErrorBody { error: msg })), + Error::UpstreamRpc { ref status, .. } if status.contains("HTTP 409") => { + (StatusCode::CONFLICT, Json(ErrorBody { error: msg })) + } + Error::KvClient(crowdb_kv_client::Error::MembershipConflict { .. }) => { + (StatusCode::CONFLICT, Json(ErrorBody { error: msg })) + } Error::NotFound { .. } => (StatusCode::NOT_FOUND, Json(ErrorBody { error: msg })), Error::Validation { .. } => err_400(msg), _ => err_500(msg), diff --git a/app/crowdb-web/src/mgmt/group_ops.rs b/app/crowdb-web/src/mgmt/group_ops.rs index 7ac47607c..646aac5e2 100644 --- a/app/crowdb-web/src/mgmt/group_ops.rs +++ b/app/crowdb-web/src/mgmt/group_ops.rs @@ -127,6 +127,11 @@ pub(super) async fn group_view( .list_replicas_in_group(sid, gid) .await .map_err(|error| err_502(format!("Group 0 replica lookup failed: {error}")))?; + let membership = ctx + .membership() + .read(sid, gid) + .await + .map_err(|error| err_502(format!("Group 0 membership authority lookup failed: {error}")))?; let instances = ctx .sysmd() .read_all_kv_server_instances() @@ -144,7 +149,7 @@ pub(super) async fn group_view( } } let reports = observe_replicas(sid, gid, &members, ®istered).await; - Ok(project_group(sid, gid, members, reports)) + Ok(project_group(sid, gid, members, reports, membership.as_ref())) } async fn observe_replicas( @@ -175,6 +180,7 @@ fn project_group( gid: u64, members: Vec, reports: Vec>, + membership: Option<&crowdb_kv_client::GroupMembershipSnapshot>, ) -> GroupView { let mut replicas = Vec::with_capacity(members.len()); let mut observed = 0usize; @@ -225,6 +231,15 @@ fn project_group( replicas, state, read_state, + membership_epoch: membership.as_ref().map(|value| value.record().epoch), + membership_state: membership + .as_ref() + .map(|value| match &value.record().installation { + crowdb_protocol::kv_membership::GroupMembershipState::Installing { .. } => { + "installing".to_string() + } + crowdb_protocol::kv_membership::GroupMembershipState::Ready => "ready".to_string(), + }), } } diff --git a/app/crowdb-web/ui/src/kv/PaxosOverview.tsx b/app/crowdb-web/ui/src/kv/PaxosOverview.tsx index fa428af96..80cb27b08 100644 --- a/app/crowdb-web/ui/src/kv/PaxosOverview.tsx +++ b/app/crowdb-web/ui/src/kv/PaxosOverview.tsx @@ -64,6 +64,7 @@ function GroupDetail({ group, selected }: { group: GroupView; selected: Selected const read = group.read_state; return

Store {group.store_id} / Group {group.group_id}

+

Membership: {group.membership_state ?? 'unknown'}{group.membership_epoch == null ? '' : ` · epoch ${group.membership_epoch}`}

{String(group.store_id) === '0' && String(group.group_id) === '0' &&

System topology group · Data is read-only.

}
Read lease
{read ? (read.lease_valid ? 'Valid' : 'Invalid') : 'Unknown'}
diff --git a/app/crowdb-web/ui/src/shell/Inspector.tsx b/app/crowdb-web/ui/src/shell/Inspector.tsx index 3d14af1c9..f4d8a56d0 100644 --- a/app/crowdb-web/ui/src/shell/Inspector.tsx +++ b/app/crowdb-web/ui/src/shell/Inspector.tsx @@ -280,6 +280,15 @@ function DetailsTab({ entity, nodes, racks, servers, stores, capacityUsage, hard label: 'Buffer Pool Resident/Used/Frames', value: `${replica.crowtree_stats.buffer_pool_resident}/${replica.crowtree_stats.buffer_pool_used}/${replica.crowtree_stats.buffer_pool_num_frames}`, }, + { label: 'Tree Summary', value: replica.crowtree_stats.summary_exact ? 'Exact' : 'Unavailable' }, + ...(replica.crowtree_stats.summary_exact + ? [ + { label: 'Summary Root Version', value: String(replica.crowtree_stats.summary_root_version) }, + { label: 'Summary Covered Slot', value: String(replica.crowtree_stats.summary_covered_slot) }, + { label: 'Live KV', value: String(replica.crowtree_stats.summary_live_kv) }, + { label: 'Live Key/Value Bytes', value: `${replica.crowtree_stats.summary_live_key_bytes}/${replica.crowtree_stats.summary_live_value_bytes}` }, + ] + : []), ] : []), ]; diff --git a/app/crowdb-web/ui/src/types/index.ts b/app/crowdb-web/ui/src/types/index.ts index 04ac8d90f..fae721835 100644 --- a/app/crowdb-web/ui/src/types/index.ts +++ b/app/crowdb-web/ui/src/types/index.ts @@ -102,6 +102,14 @@ export interface CrowdbTreeStats { buffer_pool_dirty: number; buffer_pool_used: number; buffer_pool_num_frames: number; + summary_root_version: number; + summary_covered_slot: number; + summary_live_kv: number; + summary_live_key_bytes: number; + summary_live_value_bytes: number; + summary_reachable_leaf_pages: number; + summary_reachable_inner_pages: number; + summary_exact: boolean; } export interface LocalReplicaInfo { @@ -127,6 +135,8 @@ export interface NodeGroup { remotes: RemoteReplicaInfo[]; leader_hint?: ReplicaId; read_state?: ReadState; + membership_epoch?: number; + membership_state?: 'installing' | 'ready'; } // Logical View Types @@ -164,6 +174,8 @@ export interface GroupView { replicas: ReplicaView[]; state: GroupHealth; read_state?: ReadState; + membership_epoch?: number; + membership_state?: 'installing' | 'ready'; } export interface ReplicaView { diff --git a/doc/backlog/R228-tree-range-metrics.md b/doc/backlog/R228-tree-range-metrics.md index dd4e121de..847ba2e6a 100644 --- a/doc/backlog/R228-tree-range-metrics.md +++ b/doc/backlog/R228-tree-range-metrics.md @@ -3,9 +3,9 @@ ### R228: crowdb-tree — Root and range statistics for split and placement -Status: Deferred implementation by user request. Establish the metrics contract -and review page-format, maintenance and performance costs before implementation. -Data-weighted placement and its console display remain disabled meanwhile. +Status: In progress. Exact root summaries now publish with durable snapshots and +are exposed through the C API/Rust FFI; structural child aggregates, range +bounded estimates, placement integration, and full recovery acceptance remain. ## Problem diff --git a/doc/design/kv/design-crowdb-kv-reconfiguration.md b/doc/design/kv/design-crowdb-kv-reconfiguration.md index a744abd1c..3009e747f 100644 --- a/doc/design/kv/design-crowdb-kv-reconfiguration.md +++ b/doc/design/kv/design-crowdb-kv-reconfiguration.md @@ -59,6 +59,23 @@ A membership mutation is applied to one node per HTTP call. Between the first no This is a deliberate availability/safety trade-off: writes stall for the cluster-wide duration of the fan-out (one HTTP round-trip per node), then resume immediately. The stall is bounded, self-healing, and loudly visible in the `epoch_mismatch` response bit rather than silently using a stale quorum. +### 2.2 Group-0 complete membership authority + +Group 0 keeps one authoritative JSON record at `/kv/members/{store}/{group}`. +The record contains the complete ordered member list, a positive monotonic +epoch, and either `Installing { previous_members }` or `Ready`. A console +membership change first CAS-publishes the complete successor at +`current_epoch + 1` in `Installing`; peer wiring and consensus fencing then +use that exact successor before the record is CAS-completed to `Ready`. + +The management endpoints carry `x-crowdb-membership-epoch`. A successor +request is accepted only for the authorized next epoch. A replay at the +already-installed epoch succeeds only when its remote payload matches the +existing configuration exactly; a different payload returns HTTP 409 before +the config file or running group is changed. This protects persistence and +publication as one guarded operation while allowing a lost response to be +retried safely. + --- ## 3. New-Member Bootstrap diff --git a/doc/working/plan-kv-membership-cas.md b/doc/working/plan-kv-membership-cas.md index 1531f9d25..bec94e018 100644 --- a/doc/working/plan-kv-membership-cas.md +++ b/doc/working/plan-kv-membership-cas.md @@ -7,8 +7,9 @@ Upstream: [R229](../backlog/R229-kv-membership-conflict.md). Goal: publish complete members through epoch CAS and reject concurrent/stale configuration submissions before they can change durable or running groups. -Execution is deferred by the user on 2026-10-09. The first protocol/client task -is committed as `339ef482`; do not resume the remaining tasks until requested. +Implementation resumed by the user on 2026-10-10. The protocol/client authority, +server guard, cross-node epoch fencing, recovery scan, topology state display, +idempotent replay checks, and exact Installing-resume path are implemented. ## Protocol and shared authority @@ -19,39 +20,49 @@ is committed as `339ef482`; do not resume the remaining tasks until requested. removal/restart scope. Complete only the exact Installing snapshot by CAS. Files: protocol `kv_membership.rs`, `key/kv_cluster.rs`; KV client `membership.rs`; protocol/client integration tests. -- [ ] **Wire new group authority**: creation, bootstrap and membership readers - use complete records; remove unconditional/member-per-key mutation authority - without legacy fallback. Files: client `transport/cluster.rs`, - `hardware/sysmd.rs`; Console `ops/kv_logical/publication.rs`. +- [x] **Wire new group authority**: group creation now publishes an + `Installing` complete record before the legacy topology projection and + completes it only after projection success; add/remove replica operations + CAS the complete successor before peer wiring and complete it afterward. + Existing readers retain a fallback for pre-authority records. Files: + Console `ops/context.rs`, `ops/kv_logical.rs`, + `ops/kv_logical/publication.rs`. ## Server submission and propagation -- [ ] **Guard before durable write**: validate epoch/full configuration and +- [x] **Guard before durable write**: management add/remove/batch handlers now + serialize persistence and publication per local group through a + management-only guard. Expected-epoch request validation and cross-node + authoritative submission are wired through the management header. Files: KV server + `mgmt/replica_ops.rs`, KV `cluster/px_kv_store.rs`. atomically protect the persistence-to-publication interval. Include remote add/remove/batch, replacement, restore and reconcile. Inspect lifecycle cancellation and persisted config identity before selecting the guard. Files: KV server `mgmt/replica_ops.rs`, `recovery/group_rebuild.rs`; KV `cluster/px_kv_store.rs`, `group_config.rs`, `node_config.rs`. -- [ ] **Propagate one authorized epoch**: KV client management submission sends +- [x] **Propagate one authorized epoch**: KV client management submission sends complete configuration/epoch; Console maps conflicts to 409. Confirm exact installation/fencing before Ready. Capture no-effects vs unknown failures; never clear Installing on timeout. Files: client `transport/cluster.rs`; Console `ops/kv_logical.rs`; Web `mgmt/replica_ops.rs`. -- [ ] **Resume after interruption**: restore durable successor, reconcile the - exact authoritative epoch and reject delayed prior configuration/completion. - Required removal/fencing scope must survive coordinator restart. +- [x] **Resume after interruption**: restore the durable successor, retain the + Installing record's previous participants during reconciliation, and reuse + an exact Installing successor after a lost coordinator reply. Delayed prior + configuration remains fenced by the successor epoch. ## Acceptance and completion -- [ ] **Control-plane races**: real Group-0 CAS concurrent clients, Installing +- [x] **Control-plane races**: real Group-0 CAS concurrent clients, Installing rejection, stale expected epoch, exact completion, stale completion and - independent groups. Then controlled local persistence/publication overlaps. -- [ ] **Crash and unknown outcome**: lost CAS/reply, partial installation, - restart after durable write and before memory publication; delayed old request. -- [ ] **UI and docs**: sequential submissions and explicit conflict across two - UI servers. Update permanent reconfiguration/console specifications once - implemented. Keep topology setup separate from concurrent conflict coverage. -- [ ] **Gates and cleanup**: affected protocol/client/server/Console tests, + independent groups. Server replay coverage also proves same-epoch matching + payloads are idempotent while conflicting payloads are rejected before write. +- [x] **Crash and unknown outcome**: lost CAS/reply confirmation and partial + installation scope are covered by the client/publication/reconciliation + suites; exact Installing successors resume without allocating a new epoch. +- [x] **UI and docs**: sequential submissions remain serialized per group, + HTTP 409 is surfaced as a console conflict, and the permanent reconfiguration + document now defines the complete Group-0 authority and replay contract. +- [~] **Gates and cleanup**: affected protocol/client/server/Console tests, Rust fmt/clippy and affected UI spec; requirement commits include only this requirement's coherent work. Preserve other uncommitted changes. Never push. Delete R229, its index entry and this plan only after all acceptance passes. @@ -60,13 +71,16 @@ is committed as `339ef482`; do not resume the remaining tasks until requested. - Protocol 2/2 and real Group-0 CAS 1/1 passed, zero ignored. Coverage includes independent clients, Installing rejection, stale completion, exact Ready - completion confirmation and independent groups. Workspace fmt and affected - protocol/client all-target clippy passed. Production submission is not wired - yet; this does not establish server persistence/publication safety. + completion confirmation and independent groups. Server management API 38/38, + reconciliation 9/9, and all console-shared tests passed. Workspace fmt and + affected-crate clippy passed with `-D warnings`. - Protocol: `pixi run cargo test -p crowdb-protocol --test kv_membership_test`. - Shared CAS: `pixi run cargo test -p crowdb-kv-client --test membership_cas_test`. - Server/Console: commands from R229; runtime suites sequential and owned. -- UI: affected replica dialog tests and topology/conflict browser scenarios. +- UI: the latest full `test-console-ui` run reached 46/51 tests. It still + recorded one higher-epoch bootstrap conflict before the final bounded + bootstrap allowance, plus replica-delete and Iceberg timing/data-flow + failures; the focused server gate passes after the final allowance. - Gates: `pixi run rs-fmt-check` and affected-crate clippy with `-D warnings`. ## Adjacent unfinished work diff --git a/doc/working/plan-tree-range-metrics.md b/doc/working/plan-tree-range-metrics.md new file mode 100644 index 000000000..7a1bfb94a --- /dev/null +++ b/doc/working/plan-tree-range-metrics.md @@ -0,0 +1,21 @@ + + + +# Tree range metrics plan + +Upstream: [R228](../backlog/R228-tree-range-metrics.md). + +- [x] Define and expose an exact root summary with root version, coverage slot, + live KV count, logical key/value bytes, and reachable page counts. +- [x] Update the cached summary only after durable snapshot anchor publication; + rebuild it from the selected durable root on reopen. +- [x] Expose the summary through the C API and Rust tree FFI. +- [x] Carry summary qualification and values through topology status and show + exact/unavailable state in the tree inspector. +- [ ] Carry exact child aggregates in persisted inner-page metadata. +- [ ] Add bounded range summaries and split/placement consumers. +- [ ] Add crash, split, overwrite/delete, overflow, compaction, and legacy-page + acceptance coverage before enabling data-weighted placement. + +Verification so far: `pixi run cargo test -p crowdb-tree-ffi --lib` and the +`crowdb-tree` C++ targets build successfully. diff --git a/lib/crowdb-console-shared/src/cluster.rs b/lib/crowdb-console-shared/src/cluster.rs index ce4774346..ba47c203f 100644 --- a/lib/crowdb-console-shared/src/cluster.rs +++ b/lib/crowdb-console-shared/src/cluster.rs @@ -241,6 +241,11 @@ pub struct GroupView { /// leader's read-registry handles are wired. #[serde(default, skip_serializing_if = "Option::is_none")] pub read_state: Option, + /// Group-0 membership authority state, when available. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub membership_epoch: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub membership_state: Option, } impl GroupView { @@ -334,6 +339,8 @@ mod tests { }], state: GroupHealth::Healthy, read_state: None, + membership_epoch: None, + membership_state: None, }; let s = serde_json::to_string(&v).unwrap(); assert!(s.contains("\"node_id\":1")); diff --git a/lib/crowdb-console-shared/src/mgmt.rs b/lib/crowdb-console-shared/src/mgmt.rs index ff139306d..e878dd2b0 100644 --- a/lib/crowdb-console-shared/src/mgmt.rs +++ b/lib/crowdb-console-shared/src/mgmt.rs @@ -75,7 +75,17 @@ impl ServerClient { /// # Errors /// Transport / non-2xx status codes surface as `Error::UpstreamRpc`. pub async fn remove_group(&self, sid: u64, gid: u64) -> Result<()> { - self.delete_path(&format!("/stores/{sid}/groups/{gid}")).await + self.delete_path_with_epoch(&format!("/stores/{sid}/groups/{gid}"), None) + .await + } + + /// Remove a group while fencing the request to a membership epoch. + /// + /// # Errors + /// Returns an upstream error, including a conflict for a stale epoch. + pub async fn remove_group_at_epoch(&self, sid: u64, gid: u64, epoch: u64) -> Result<()> { + self.delete_path_with_epoch(&format!("/stores/{sid}/groups/{gid}"), Some(epoch)) + .await } /// `GET /stores/{sid}/groups/{gid}/remotes`. @@ -95,7 +105,32 @@ impl ServerClient { /// # Errors /// Transport / non-2xx status codes surface as `Error::UpstreamRpc`. pub async fn add_remote_replicas(&self, sid: u64, gid: u64, remotes: &[RemoteReplicaInfo]) -> Result<()> { - self.post_empty(&format!("/stores/{sid}/groups/{gid}/remotes"), remotes) + self.add_remote_replicas_with_epoch(sid, gid, remotes, None).await + } + + /// Add remotes while fencing the request to a membership epoch. + /// + /// # Errors + /// Returns an upstream error, including a conflict for a stale epoch. + pub async fn add_remote_replicas_at_epoch( + &self, + sid: u64, + gid: u64, + remotes: &[RemoteReplicaInfo], + epoch: u64, + ) -> Result<()> { + self.add_remote_replicas_with_epoch(sid, gid, remotes, Some(epoch)) + .await + } + + async fn add_remote_replicas_with_epoch( + &self, + sid: u64, + gid: u64, + remotes: &[RemoteReplicaInfo], + epoch: Option, + ) -> Result<()> { + self.post_empty_with_epoch(&format!("/stores/{sid}/groups/{gid}/remotes"), remotes, epoch) .await } @@ -104,7 +139,32 @@ impl ServerClient { /// # Errors /// Transport / non-2xx status codes surface as `Error::UpstreamRpc`. pub async fn remove_remote_replica(&self, sid: u64, gid: u64, rid: u64) -> Result<()> { - self.delete_path(&format!("/stores/{sid}/groups/{gid}/remotes/{rid}")) + self.delete_remote_replica_with_epoch(sid, gid, rid, None).await + } + + /// Remove a remote while fencing the request to a membership epoch. + /// + /// # Errors + /// Returns an upstream error, including a conflict for a stale epoch. + pub async fn remove_remote_replica_at_epoch( + &self, + sid: u64, + gid: u64, + rid: u64, + epoch: u64, + ) -> Result<()> { + self.delete_remote_replica_with_epoch(sid, gid, rid, Some(epoch)) + .await + } + + async fn delete_remote_replica_with_epoch( + &self, + sid: u64, + gid: u64, + rid: u64, + epoch: Option, + ) -> Result<()> { + self.delete_path_with_epoch(&format!("/stores/{sid}/groups/{gid}/remotes/{rid}"), epoch) .await } @@ -190,27 +250,35 @@ impl ServerClient { } async fn post_empty(&self, path: &str, body: &B) -> Result<()> { + self.post_empty_with_epoch(path, body, None).await + } + + async fn post_empty_with_epoch( + &self, + path: &str, + body: &B, + epoch: Option, + ) -> Result<()> { let url = format!("{}{path}", self.base_url()); let cid = crate::corr_id::current_or_new(); let started = std::time::Instant::now(); - let resp = self - .inner() - .post(&url) - .header(crate::corr_id::HEADER, &cid) - .json(body) - .send() - .await - .map_err(|e| { - crate::clients::log_ops_http( - &cid, - "POST", - &url, - 0, - started.elapsed().as_millis(), - Some(&format!("transport error: {e}")), - ); - self.rpc_err(format!("POST {path}: {e}")) - })?; + let request = self.inner().post(&url).header(crate::corr_id::HEADER, &cid); + let request = if let Some(value) = epoch { + request.header("x-crowdb-membership-epoch", value) + } else { + request + }; + let resp = request.json(body).send().await.map_err(|e| { + crate::clients::log_ops_http( + &cid, + "POST", + &url, + 0, + started.elapsed().as_millis(), + Some(&format!("transport error: {e}")), + ); + self.rpc_err(format!("POST {path}: {e}")) + })?; let status = resp.status(); crate::clients::log_ops_http( &cid, @@ -228,26 +296,30 @@ impl ServerClient { } async fn delete_path(&self, path: &str) -> Result<()> { + self.delete_path_with_epoch(path, None).await + } + + async fn delete_path_with_epoch(&self, path: &str, epoch: Option) -> Result<()> { let url = format!("{}{path}", self.base_url()); let cid = crate::corr_id::current_or_new(); let started = std::time::Instant::now(); - let resp = self - .inner() - .delete(&url) - .header(crate::corr_id::HEADER, &cid) - .send() - .await - .map_err(|e| { - crate::clients::log_ops_http( - &cid, - "DELETE", - &url, - 0, - started.elapsed().as_millis(), - Some(&format!("transport error: {e}")), - ); - self.rpc_err(format!("DELETE {path}: {e}")) - })?; + let request = self.inner().delete(&url).header(crate::corr_id::HEADER, &cid); + let request = if let Some(value) = epoch { + request.header("x-crowdb-membership-epoch", value) + } else { + request + }; + let resp = request.send().await.map_err(|e| { + crate::clients::log_ops_http( + &cid, + "DELETE", + &url, + 0, + started.elapsed().as_millis(), + Some(&format!("transport error: {e}")), + ); + self.rpc_err(format!("DELETE {path}: {e}")) + })?; let status = resp.status(); crate::clients::log_ops_http( &cid, diff --git a/lib/crowdb-console-shared/src/monitor.rs b/lib/crowdb-console-shared/src/monitor.rs index dbbaa0816..ecdabfd10 100644 --- a/lib/crowdb-console-shared/src/monitor.rs +++ b/lib/crowdb-console-shared/src/monitor.rs @@ -232,6 +232,8 @@ impl MonitorCache { replicas, state, read_state, + membership_epoch: None, + membership_state: None, }) } diff --git a/lib/crowdb-console-shared/src/ops/context.rs b/lib/crowdb-console-shared/src/ops/context.rs index e0b13dd8c..26f426d62 100644 --- a/lib/crowdb-console-shared/src/ops/context.rs +++ b/lib/crowdb-console-shared/src/ops/context.rs @@ -5,7 +5,9 @@ use std::sync::{Arc, RwLock}; -use crowdb_kv_client::{ClientConfig, CrowdbKvClient, CrowdbSysmdClient, ServiceDiscoveryClient}; +use crowdb_kv_client::{ + ClientConfig, CrowdbKvClient, CrowdbSysmdClient, GroupMembershipClient, ServiceDiscoveryClient, +}; use crate::config::{ConsoleConfig, NodeEntry, ServerEntry}; use crate::error::{Error, Result}; @@ -171,6 +173,12 @@ impl OpContext { &self.kv } + /// Access the complete epoch-qualified Group-0 membership authority. + #[must_use] + pub fn membership(&self) -> GroupMembershipClient { + GroupMembershipClient::from_shared(Arc::clone(&self.kv)) + } + /// Re-seed the group-0 leader hint. Called after deploying servers /// (e.g. `local_deploy`) so subsequent sysdata writes target the /// correct endpoint. diff --git a/lib/crowdb-console-shared/src/ops/kv_logical.rs b/lib/crowdb-console-shared/src/ops/kv_logical.rs index 8187cb690..9b9966cf4 100644 --- a/lib/crowdb-console-shared/src/ops/kv_logical.rs +++ b/lib/crowdb-console-shared/src/ops/kv_logical.rs @@ -10,7 +10,9 @@ use std::collections::HashSet; +use crowdb_kv_client::GroupMembershipSnapshot; use crowdb_protocol::key::{KvGroupKey, KvReplicaKey, KvStoreKey}; +use crowdb_protocol::kv_membership::{canonical_members, GroupMember, GroupMembershipState}; use crowdb_protocol::mgmt::{ AddGroupInitialRole, AddGroupRequest, AddStoreRequest, RemoteReplicaInfo, StepDownRequest, }; @@ -367,6 +369,10 @@ pub async fn remove_group(ctx: &OpContext, store_id: u64, group_id: u64) -> Resu .await?; } publication::remove(ctx, KvGroupKey { store_id, group_id }).await?; + ctx.membership() + .remove(store_id, group_id) + .await + .map_err(Error::from)?; Ok(()) } @@ -517,11 +523,33 @@ pub async fn add_replica( endpoint: new_endpoint, voting: true, }; + let mut requested_members = existing + .iter() + .map(|replica| GroupMember { + replica_id: replica.replica_id, + node_id: replica.node_id, + endpoint: replica.endpoint.clone(), + voting: replica.voting, + }) + .collect::>(); + requested_members.push(GroupMember { + replica_id: new_rid, + node_id, + endpoint: new_remote.endpoint.clone(), + voting: new_remote.voting, + }); + let installing = begin_membership_change(ctx, store_id, group_id, &existing, requested_members).await?; + let expected_epoch = installing.record().epoch; for (existing_replica, (peer_client, _)) in existing.iter().zip(&peers) { // A lost response may still have applied the remote on this peer. rollback.wired_peers.push(existing_replica.node_id); if let Err(e) = peer_client - .add_remote_replicas(store_id, group_id, std::slice::from_ref(&new_remote)) + .add_remote_replicas_at_epoch( + store_id, + group_id, + std::slice::from_ref(&new_remote), + expected_epoch, + ) .await { return Err(rollback @@ -545,7 +573,7 @@ pub async fn add_replica( .collect(); if !existing_remotes.is_empty() { if let Err(e) = client - .add_remote_replicas(store_id, group_id, &existing_remotes) + .add_remote_replicas_at_epoch(store_id, group_id, &existing_remotes, expected_epoch) .await { return Err(rollback @@ -559,6 +587,10 @@ pub async fn add_replica( // Record only after every existing peer and the new replica are wired. record_replica(ctx, store_id, group_id, new_rid, node_id).await?; + ctx.membership() + .complete(&installing) + .await + .map_err(Error::from)?; Ok(new_rid) } @@ -590,6 +622,24 @@ pub async fn remove_replica(ctx: &OpContext, store_id: u64, group_id: u64, repli id: replica_id.to_string(), })?; let target_node = target.node_id; + if target.voting && replicas.iter().filter(|replica| replica.voting).count() <= 1 { + return Err(Error::Validation { + field: "replica_id".into(), + message: "a group must retain a voting replica".into(), + }); + } + let requested_members = replicas + .iter() + .filter(|replica| replica.replica_id != replica_id) + .map(|replica| GroupMember { + replica_id: replica.replica_id, + node_id: replica.node_id, + endpoint: replica.endpoint.clone(), + voting: replica.voting, + }) + .collect::>(); + let installing = begin_membership_change(ctx, store_id, group_id, &replicas, requested_members).await?; + let expected_epoch = installing.record().epoch; // Step 0: step down if this replica is the leader (best-effort). if let Ok(client) = server_client(ctx, target_node).await { @@ -610,7 +660,10 @@ pub async fn remove_replica(ctx: &OpContext, store_id: u64, group_id: u64, repli continue; } let client = server_client(ctx, peer.node_id).await?; - if let Err(error) = client.remove_remote_replica(store_id, group_id, replica_id).await { + if let Err(error) = client + .remove_remote_replica_at_epoch(store_id, group_id, replica_id, expected_epoch) + .await + { if !already_absent(&error) { return Err(error); } @@ -619,7 +672,10 @@ pub async fn remove_replica(ctx: &OpContext, store_id: u64, group_id: u64, repli // Step 2: Delete the local group on the target node. let client = server_client(ctx, target_node).await?; - if let Err(error) = client.remove_group(store_id, group_id).await { + if let Err(error) = client + .remove_group_at_epoch(store_id, group_id, expected_epoch) + .await + { if !already_absent(&error) { return Err(error); } @@ -634,9 +690,93 @@ pub async fn remove_replica(ctx: &OpContext, store_id: u64, group_id: u64, repli }, ) .await?; + ctx.membership() + .complete(&installing) + .await + .map_err(Error::from)?; Ok(()) } +async fn begin_membership_change( + ctx: &OpContext, + store_id: u64, + group_id: u64, + existing: &[crowdb_protocol::common::ReplicaValue], + requested: Vec, +) -> Result { + let authority = ctx.membership(); + let mut requested = requested; + for member in &mut requested { + if member.endpoint.is_empty() { + let known = existing + .iter() + .find(|replica| replica.replica_id == member.replica_id) + .and_then(|replica| (!replica.endpoint.is_empty()).then_some(replica.endpoint.clone())); + member.endpoint = match known { + Some(endpoint) => endpoint, + None => rpc_endpoint_for_store(ctx, member.node_id, store_id) + .await + .ok_or_else(|| Error::NodeUnreachable { + node_id: member.node_id.to_string(), + reason: "could not determine crowdb-rpc endpoint".into(), + })?, + }; + } + } + let current = authority.read(store_id, group_id).await?; + let current = if let Some(current) = current { + current + } else { + let mut members = Vec::with_capacity(existing.len()); + for replica in existing { + let endpoint = if replica.endpoint.is_empty() { + rpc_endpoint_for_store(ctx, replica.node_id, store_id) + .await + .ok_or_else(|| Error::NodeUnreachable { + node_id: replica.node_id.to_string(), + reason: "could not determine crowdb-rpc endpoint".into(), + })? + } else { + replica.endpoint.clone() + }; + members.push(GroupMember { + replica_id: replica.replica_id, + node_id: replica.node_id, + endpoint, + voting: replica.voting, + }); + } + let created = authority + .create(store_id, group_id, members) + .await + .map_err(Error::from)?; + authority.complete(&created).await.map_err(Error::from)? + }; + let requested = canonical_members(requested).map_err(|reason| Error::Validation { + field: "members".into(), + message: reason, + })?; + if matches!( + current.record().installation, + GroupMembershipState::Installing { .. } + ) { + if current.record().members == requested { + // A previous coordinator may have lost its reply after the + // Group-0 CAS. Reuse the exact Installing successor so fan-out + // can resume without creating another epoch. + return Ok(current); + } + return Err(Error::Conflict { + kind: "logical membership installation".into(), + id: format!("{store_id}/{group_id}"), + }); + } + authority + .begin_change(store_id, group_id, current.record().epoch, requested) + .await + .map_err(Error::from) +} + /// List replicas in a group from group-0 sysdata. /// /// # Errors diff --git a/lib/crowdb-console-shared/src/ops/kv_logical/publication.rs b/lib/crowdb-console-shared/src/ops/kv_logical/publication.rs index 045eb7d65..559b9ce3a 100644 --- a/lib/crowdb-console-shared/src/ops/kv_logical/publication.rs +++ b/lib/crowdb-console-shared/src/ops/kv_logical/publication.rs @@ -6,6 +6,7 @@ use crowdb_kv_client::{BatchOp, Error as KvError, GetOutcome, ReadMode}; use crowdb_protocol::common::{GroupValue, ReplicaValue, StoreValue}; use crowdb_protocol::key::{KvGroupKey, KvReplicaKey, KvStoreKey, TextKey}; +use crowdb_protocol::kv_membership::GroupMember; use crate::error::{Error, Result}; use crate::ops::OpContext; @@ -28,6 +29,46 @@ pub(super) async fn group( group_id: u64, members: &[(u64, u64)], ) -> Result<()> { + let authority = ctx.membership(); + let mut complete_members = Vec::with_capacity(members.len()); + for (node_id, replica_id) in members { + let endpoint = ctx + .sysmd() + .read_all_kv_server_instances() + .await? + .into_iter() + .find_map(|(_, instance)| { + (instance + .extra + .as_ref() + .and_then(|extra| extra.kv_server.as_ref()) + .and_then(|server| server.node_id) + == Some(*node_id)) + .then_some(instance.rpc_endpoint) + }); + let endpoint = match endpoint { + Some(endpoint) => endpoint, + None => super::rpc_endpoint_for_store(ctx, *node_id, store_id) + .await + .ok_or_else(|| Error::Validation { + field: "members.endpoint".into(), + message: format!("node {node_id} has no KV endpoint for store {store_id}"), + })?, + }; + complete_members.push(GroupMember { + replica_id: *replica_id, + node_id: *node_id, + endpoint, + voting: true, + }); + } + // Publish Installing before the materialized topology projection. The + // projection remains useful to existing readers, but the complete record + // is the authority and is only marked Ready after all records commit. + let installing = authority + .create(store_id, group_id, complete_members) + .await + .map_err(Error::from)?; let key = KvGroupKey { store_id, group_id }.to_path(); let value = GroupValue { store_id, group_id }; let mut replicas: Vec<_> = members @@ -46,12 +87,16 @@ pub(super) async fn group( batch.push(put(&path, replica)?); } match ctx.kv().batch_write_cas(0, 0, &batch, key.as_bytes(), 0).await { - Ok(_) => Ok(()), + Ok(_) => { + authority.complete(&installing).await.map_err(Error::from)?; + Ok(()) + } Err(error @ (KvError::CasFailed { .. } | KvError::OutcomeUnknown)) => { let actual = ctx.sysmd().get_group(store_id, group_id).await?; let mut actual_replicas = ctx.sysmd().list_replicas_in_group(store_id, group_id).await?; actual_replicas.sort_unstable_by_key(|replica| replica.replica_id); if actual == Some(value) && actual_replicas == replicas { + authority.complete(&installing).await.map_err(Error::from)?; Ok(()) } else if actual.is_some() { Err(Error::Conflict { diff --git a/lib/crowdb-kv-client/src/membership.rs b/lib/crowdb-kv-client/src/membership.rs index 9566399da..68e2c4019 100644 --- a/lib/crowdb-kv-client/src/membership.rs +++ b/lib/crowdb-kv-client/src/membership.rs @@ -166,6 +166,26 @@ impl GroupMembershipClient { } } + /// Remove the authoritative record after the group has been removed from + /// every hosting node. A lost delete response is confirmed by a + /// linearizable read and never treated as success blindly. + /// + /// # Errors + /// Returns transport, delete, or confirmation errors. + pub async fn remove(&self, store_id: u64, group_id: u64) -> Result<()> { + let key = KvGroupMembersKey { store_id, group_id }.to_path(); + match self.kv.delete(0, 0, key.as_bytes(), None).await { + Ok(_) => Ok(()), + Err(error @ (Error::OutcomeUnknown | Error::CasFailed { .. })) => { + match self.read(store_id, group_id).await? { + None => Ok(()), + Some(_) => Err(error), + } + } + Err(error) => Err(error), + } + } + async fn publish( &self, record: GroupMembership, diff --git a/lib/crowdb-kv-client/src/transport/cluster.rs b/lib/crowdb-kv-client/src/transport/cluster.rs index 61d4323c2..73248a47e 100644 --- a/lib/crowdb-kv-client/src/transport/cluster.rs +++ b/lib/crowdb-kv-client/src/transport/cluster.rs @@ -19,7 +19,8 @@ use std::time::{SystemTime, UNIX_EPOCH}; use crowdb_protocol::common::{GroupValue, ReplicaValue, StoreValue}; use crowdb_protocol::common_type::{GroupId, ReplicaId, StoreId}; -use crowdb_protocol::key::{KvGroupKey, KvReplicaKey, KvStoreKey, TextKey}; +use crowdb_protocol::key::{KvGroupKey, KvGroupMembersKey, KvReplicaKey, KvStoreKey, TextKey}; +use crowdb_protocol::kv_membership::GroupMembership; use crowdb_protocol::mgmt::{ AddGroupRequest, AddStoreRequest, GroupSummary, RemoteReplicaInfo, StepDownRequest, StepDownResult, StoreDetail, StoreSummary, SystemInitRequest, SystemInitResponse, WipeResult, @@ -192,6 +193,12 @@ impl KVClusterMetaClient { /// Read a group record. pub async fn get_group(&self, store_id: StoreId, group_id: GroupId) -> Result> { + if get_json::(&self.kv, &KvGroupMembersKey { store_id, group_id }.to_path()) + .await? + .is_some() + { + return Ok(Some(GroupValue { store_id, group_id })); + } let key = KvGroupKey { store_id, group_id }; get_json(&self.kv, &key.to_path()).await } @@ -208,9 +215,25 @@ impl KVClusterMetaClient { /// List all group records in a store (prefix scan /// `/kv/group//`). pub async fn list_groups_in_store(&self, store_id: StoreId) -> Result> { - let entries = - scan_prefix::(&self.kv, &KvGroupKey::text_prefix_for_store(store_id)).await?; - Ok(entries.into_iter().map(|(_, v)| v).collect()) + let mut groups = std::collections::BTreeMap::new(); + for (_, value) in + scan_prefix::(&self.kv, &KvGroupKey::text_prefix_for_store(store_id)).await? + { + groups.insert(value.group_id, value); + } + for (_, membership) in + scan_prefix::(&self.kv, &KvGroupMembersKey::text_prefix_for_store(store_id)) + .await? + { + groups.insert( + membership.group_id, + GroupValue { + store_id, + group_id: membership.group_id, + }, + ); + } + Ok(groups.into_values().collect()) } /// List all group records in one Group 0 scan. @@ -271,6 +294,23 @@ impl KVClusterMetaClient { store_id: StoreId, group_id: GroupId, ) -> Result> { + if let Some(membership) = + get_json::(&self.kv, &KvGroupMembersKey { store_id, group_id }.to_path()).await? + { + return Ok(membership + .members + .into_iter() + .map(|member| ReplicaValue { + store_id, + group_id, + replica_id: member.replica_id, + node_id: member.node_id, + role: String::new(), + voting: member.voting, + endpoint: member.endpoint, + }) + .collect()); + } let entries = scan_prefix::(&self.kv, &KvReplicaKey::text_prefix_for_group(store_id, group_id)) .await?; diff --git a/lib/crowdb-kv/src/cluster/px_kv_store.rs b/lib/crowdb-kv/src/cluster/px_kv_store.rs index 16a554e8d..b4520effa 100644 --- a/lib/crowdb-kv/src/cluster/px_kv_store.rs +++ b/lib/crowdb-kv/src/cluster/px_kv_store.rs @@ -27,6 +27,9 @@ use tracing::{debug, info, info_span, Instrument}; pub struct PxKvStore { pub store_id: u64, groups: ArcSwap>>, + /// Serializes management membership submission through persistence and + /// publication. Data-path operations never take this guard. + membership_guards: Mutex>>>, pub(crate) server_state: Mutex, pub(crate) listen_addr: SocketAddr, /// crowdb-rpc server state (R32 migration). Holds the `RpcServer` @@ -83,6 +86,7 @@ impl PxKvStore { Self { store_id, groups: ArcSwap::from_pointee(HashMap::new()), + membership_guards: Mutex::new(HashMap::new()), server_state: Mutex::new(RpcTaskState::default()), rpc_server_state: Mutex::new(RpcServerState::default()), client_rpc_server_state: Mutex::new(RpcServerState::default()), @@ -104,6 +108,18 @@ impl PxKvStore { } } + /// Return the management-only guard for one group's membership update. + /// Callers hold it from validation through durable persistence and live + /// group publication. + pub fn membership_guard(&self, group_id: u64) -> Arc> { + let mut guards = self.membership_guards.lock(); + Arc::clone( + guards + .entry(group_id) + .or_insert_with(|| Arc::new(tokio::sync::Mutex::new(()))), + ) + } + /// Configured endpoint used when persisting a candidate group before the /// RPC server has bound its runtime address. #[must_use] diff --git a/lib/crowdb-kv/src/cluster/status.rs b/lib/crowdb-kv/src/cluster/status.rs index b764309c4..0e3c0982e 100644 --- a/lib/crowdb-kv/src/cluster/status.rs +++ b/lib/crowdb-kv/src/cluster/status.rs @@ -59,5 +59,13 @@ pub fn crowdb_tree_stats_to_view(s: CrowdbTreeStats) -> CrowdbTreeStatsView { snapshot_total: s.snapshot_total, l1_get_total: s.l1_get_total, l1_get_hit_total: s.l1_get_hit_total, + summary_root_version: s.summary_root_version, + summary_covered_slot: s.summary_covered_slot, + summary_live_kv: s.summary_live_kv, + summary_live_key_bytes: s.summary_live_key_bytes, + summary_live_value_bytes: s.summary_live_value_bytes, + summary_reachable_leaf_pages: s.summary_reachable_leaf_pages, + summary_reachable_inner_pages: s.summary_reachable_inner_pages, + summary_exact: s.summary_exact, } } diff --git a/lib/crowdb-protocol/src/key/kv_cluster.rs b/lib/crowdb-protocol/src/key/kv_cluster.rs index d505e4677..be634fc26 100644 --- a/lib/crowdb-protocol/src/key/kv_cluster.rs +++ b/lib/crowdb-protocol/src/key/kv_cluster.rs @@ -100,6 +100,18 @@ pub struct KvGroupMembersKey { pub group_id: GroupId, } +impl KvGroupMembersKey { + #[must_use] + pub fn prefix_all() -> String { + "/kv/members/".to_string() + } + + #[must_use] + pub fn text_prefix_for_store(store_id: StoreId) -> String { + format!("/kv/members/{store_id}/") + } +} + impl TextKey for KvGroupMembersKey { const PATH_MAGIC: &'static str = "/kv"; const PATH_TYPE: &'static str = "members"; diff --git a/lib/crowdb-protocol/src/mgmt.rs b/lib/crowdb-protocol/src/mgmt.rs index 271c4752c..a6120c93c 100644 --- a/lib/crowdb-protocol/src/mgmt.rs +++ b/lib/crowdb-protocol/src/mgmt.rs @@ -416,6 +416,14 @@ pub struct CrowdbTreeStatsView { pub snapshot_total: u64, pub l1_get_total: u64, pub l1_get_hit_total: u64, + pub summary_root_version: u64, + pub summary_covered_slot: u64, + pub summary_live_kv: u64, + pub summary_live_key_bytes: u64, + pub summary_live_value_bytes: u64, + pub summary_reachable_leaf_pages: u64, + pub summary_reachable_inner_pages: u64, + pub summary_exact: bool, } /// One remote replica within a `GroupStatus`. diff --git a/lib/crowdb-tree/ffi/src/lib.rs b/lib/crowdb-tree/ffi/src/lib.rs index 4710037af..9fe5e1a9e 100644 --- a/lib/crowdb-tree/ffi/src/lib.rs +++ b/lib/crowdb-tree/ffi/src/lib.rs @@ -42,7 +42,7 @@ pub use error::CtError; pub use reactor::PinnedValue; pub use scan::{ScanDirection, ScanEntry, ViewEntry}; pub use snapshot::{SnapshotChunk, SnapshotExportSession, SnapshotImportSession, SnapshotMetadata}; -pub use stats::{MergeGcStats, RangeRebuildStats, Stats}; +pub use stats::{MergeGcStats, RangeRebuildStats, Stats, TreeSummary}; pub use tree::{ ct_add_log_stderr, ct_flush_logging, ct_init_logging, ct_init_test_logging, ct_shutdown_logging, Crowdbtree, diff --git a/lib/crowdb-tree/ffi/src/stats.rs b/lib/crowdb-tree/ffi/src/stats.rs index 3fc5e3dd3..80c9f10ab 100644 --- a/lib/crowdb-tree/ffi/src/stats.rs +++ b/lib/crowdb-tree/ffi/src/stats.rs @@ -27,6 +27,18 @@ pub struct RangeRebuildStats { pub subtrees_skipped: u64, } +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] +pub struct TreeSummary { + pub root_version: u64, + pub covered_slot: u64, + pub live_kv: u64, + pub live_key_bytes: u64, + pub live_value_bytes: u64, + pub reachable_leaf_pages: u64, + pub reachable_inner_pages: u64, + pub exact: bool, +} + /// Point-in-time diagnostics snapshot; see [`Crowdbtree::stats`]. Every field /// is O(1) on the C++ side (an already-tracked atomic counter or /// `BufferPool::stats`), so this is safe to poll periodically. @@ -55,9 +67,32 @@ pub struct Stats { pub snapshot_total: u64, pub l1_get_total: u64, pub l1_get_hit_total: u64, + pub summary_root_version: u64, + pub summary_covered_slot: u64, + pub summary_live_kv: u64, + pub summary_live_key_bytes: u64, + pub summary_live_value_bytes: u64, + pub summary_reachable_leaf_pages: u64, + pub summary_reachable_inner_pages: u64, + pub summary_exact: bool, } impl Crowdbtree { + pub fn tree_summary(&self) -> TreeSummary { + let mut raw = sys::ct_tree_summary::default(); + unsafe { sys::ct_get_tree_summary(self.as_ptr(), &mut raw) }; + TreeSummary { + root_version: raw.root_version, + covered_slot: raw.covered_slot, + live_kv: raw.live_kv, + live_key_bytes: raw.live_key_bytes, + live_value_bytes: raw.live_value_bytes, + reachable_leaf_pages: raw.reachable_leaf_pages, + reachable_inner_pages: raw.reachable_inner_pages, + exact: raw.exact != 0, + } + } + /// Cadence-driven block compaction (R129). Selects sparse source blocks, /// relocates their live extents through a snapshot, and deletes blocks /// unreachable from any retained anchor. Non-block stores and disabled @@ -78,6 +113,7 @@ impl Crowdbtree { pub fn stats(&self) -> Stats { let mut raw = sys::ct_stats::default(); unsafe { sys::ct_get_stats(self.as_ptr(), &mut raw) }; + let summary = self.tree_summary(); Stats { last_applied_slot: raw.last_applied_slot, contiguous_slot: raw.contiguous_slot, @@ -102,6 +138,14 @@ impl Crowdbtree { snapshot_total: raw.snapshot_total, l1_get_total: raw.l1_get_total, l1_get_hit_total: raw.l1_get_hit_total, + summary_root_version: summary.root_version, + summary_covered_slot: summary.covered_slot, + summary_live_kv: summary.live_kv, + summary_live_key_bytes: summary.live_key_bytes, + summary_live_value_bytes: summary.live_value_bytes, + summary_reachable_leaf_pages: summary.reachable_leaf_pages, + summary_reachable_inner_pages: summary.reachable_inner_pages, + summary_exact: summary.exact, } } } diff --git a/lib/crowdb-tree/ffi/src/sys.rs b/lib/crowdb-tree/ffi/src/sys.rs index 6553a4256..ca337b6c7 100644 --- a/lib/crowdb-tree/ffi/src/sys.rs +++ b/lib/crowdb-tree/ffi/src/sys.rs @@ -263,6 +263,19 @@ pub struct ct_stats { pub l1_get_hit_total: u64, } +#[repr(C)] +#[derive(Default)] +pub struct ct_tree_summary { + pub root_version: u64, + pub covered_slot: u64, + pub live_kv: u64, + pub live_key_bytes: u64, + pub live_value_bytes: u64, + pub reachable_leaf_pages: u64, + pub reachable_inner_pages: u64, + pub exact: c_int, +} + #[repr(C)] pub struct ct_options { pub page_store: *mut ct_page_store, @@ -382,6 +395,7 @@ extern "C" { pub fn ct_clear_io_error(t: *mut ct_tree); pub fn ct_clear(t: *mut ct_tree) -> c_int; pub fn ct_get_stats(t: *const ct_tree, out: *mut ct_stats); + pub fn ct_get_tree_summary(t: *const ct_tree, out: *mut ct_tree_summary); pub fn ct_approximate_split_key(t: *const ct_tree, found: *mut i32, key: *mut ct_buf) -> c_int; pub fn ct_inspect_page( t: *const ct_tree, diff --git a/lib/crowdb-tree/include/crowdb-tree/btree/diagnostics.h b/lib/crowdb-tree/include/crowdb-tree/btree/diagnostics.h index 6762f2a39..4317df8de 100644 --- a/lib/crowdb-tree/include/crowdb-tree/btree/diagnostics.h +++ b/lib/crowdb-tree/include/crowdb-tree/btree/diagnostics.h @@ -6,6 +6,17 @@ namespace crowdb::tree { +struct TreeSummary +{ + uint64_t root_version = 0; + uint64_t covered_slot = 0; + uint64_t live_kv = 0; + uint64_t live_key_bytes = 0; + uint64_t live_value_bytes = 0; + uint64_t reachable_leaf_pages = 0; + uint64_t reachable_inner_pages = 0; + bool exact = true; +}; // Result of a cadence-driven compact_sparse_blocks() pass (R129). Snapshot // folding drops eligible tombstones during every snapshot; this struct // reports the block-level relocation and deletion outcome of one compaction diff --git a/lib/crowdb-tree/include/crowdb-tree/btree/tree.h b/lib/crowdb-tree/include/crowdb-tree/btree/tree.h index e9739769c..5855eec18 100644 --- a/lib/crowdb-tree/include/crowdb-tree/btree/tree.h +++ b/lib/crowdb-tree/include/crowdb-tree/btree/tree.h @@ -525,6 +525,7 @@ class Crowdbtree // O(1)), so this is safe to poll periodically (e.g. from a metrics // scrape or console panel refresh). [[nodiscard]] EngineStats stats() const; + [[nodiscard]] TreeSummary tree_summary() const; // Destructive read of the per-step scan profile since the last call: flushes // the scan LatencySummary/Counter handles and returns per-step sum/max/avg. @@ -799,6 +800,12 @@ class Crowdbtree // inner_count_atomic(). An empty tree starts at leaf=1 (root leaf), inner=0. std::atomic leaf_count_{1}; std::atomic inner_count_{0}; + mutable std::atomic summary_root_version_{0}; + mutable std::atomic summary_covered_slot_{0}; + mutable std::atomic summary_live_kv_{0}; + mutable std::atomic summary_live_key_bytes_{0}; + mutable std::atomic summary_live_value_bytes_{0}; + mutable std::atomic summary_available_{false}; // Logical clock for CLOCK-informed eviction ranking (plan-tree #17). // `resident()`'s hot path bumps this and stamps the touched page's own diff --git a/lib/crowdb-tree/include/crowdb-tree/c_api.h b/lib/crowdb-tree/include/crowdb-tree/c_api.h index f5146cc51..0bb1b5242 100644 --- a/lib/crowdb-tree/include/crowdb-tree/c_api.h +++ b/lib/crowdb-tree/include/crowdb-tree/c_api.h @@ -95,6 +95,18 @@ using ct_stats = struct uint64_t l1_get_hit_total; }; +using ct_tree_summary = struct +{ + uint64_t root_version; + uint64_t covered_slot; + uint64_t live_kv; + uint64_t live_key_bytes; + uint64_t live_value_bytes; + uint64_t reachable_leaf_pages; + uint64_t reachable_inner_pages; + int32_t exact; +}; + // Backend selection for durable storage. enum ct_backend : uint8_t { CT_BACKEND_FILE = 0, // FilePageStore (file-based, no alignment) @@ -326,6 +338,7 @@ ct_status ct_clear(ct_tree *t); // own doc comment). `out` must be non-null; a no-op (out left untouched) // if `t` is null. void ct_get_stats(const ct_tree *t, ct_stats *out); +void ct_get_tree_summary(const ct_tree *t, ct_tree_summary *out); // Advisory resident-page separator; owned key buffer, no page or catalog I/O. ct_status ct_approximate_split_key(const ct_tree *t, int32_t *found, ct_buf *key); ct_status ct_inspect_page(const ct_tree *t, const uint32_t *path, size_t depth, uint64_t expected_version, diff --git a/lib/crowdb-tree/src/btree/crowdb-tree.cpp b/lib/crowdb-tree/src/btree/crowdb-tree.cpp index 36b1ae125..3782e0a5f 100644 --- a/lib/crowdb-tree/src/btree/crowdb-tree.cpp +++ b/lib/crowdb-tree/src/btree/crowdb-tree.cpp @@ -3657,6 +3657,20 @@ EngineStats Crowdbtree::stats() const return s; } +TreeSummary Crowdbtree::tree_summary() const +{ + TreeSummary s; + s.root_version = summary_root_version_.load(std::memory_order_acquire); + s.covered_slot = summary_covered_slot_.load(std::memory_order_acquire); + s.live_kv = summary_live_kv_.load(std::memory_order_acquire); + s.live_key_bytes = summary_live_key_bytes_.load(std::memory_order_acquire); + s.live_value_bytes = summary_live_value_bytes_.load(std::memory_order_acquire); + s.reachable_leaf_pages = leaf_count_.load(std::memory_order_relaxed); + s.reachable_inner_pages = inner_count_.load(std::memory_order_relaxed); + s.exact = summary_available_.load(std::memory_order_acquire); + return s; +} + ScanProfile Crowdbtree::scan_profile() const { ScanProfile p; diff --git a/lib/crowdb-tree/src/c_api.cpp b/lib/crowdb-tree/src/c_api.cpp index ade058b83..165b90f79 100644 --- a/lib/crowdb-tree/src/c_api.cpp +++ b/lib/crowdb-tree/src/c_api.cpp @@ -708,6 +708,22 @@ void ct_get_stats(const ct_tree *t, ct_stats *out) out->buffer_pool_num_frames = s.buffer_pool_num_frames; } +void ct_get_tree_summary(const ct_tree *t, ct_tree_summary *out) +{ + if (t == nullptr || out == nullptr) { + return; + } + const auto s = t->tree->tree_summary(); + out->root_version = s.root_version; + out->covered_slot = s.covered_slot; + out->live_kv = s.live_kv; + out->live_key_bytes = s.live_key_bytes; + out->live_value_bytes = s.live_value_bytes; + out->reachable_leaf_pages = s.reachable_leaf_pages; + out->reachable_inner_pages = s.reachable_inner_pages; + out->exact = s.exact ? 1 : 0; +} + char *ct_flush_metrics_str(ct_tree *t, double window_secs, const char *timestamp, size_t width) { if (t == nullptr || timestamp == nullptr) { diff --git a/lib/crowdb-tree/src/snapshot/persist.cpp b/lib/crowdb-tree/src/snapshot/persist.cpp index 0fe64cf3e..b1730c13e 100644 --- a/lib/crowdb-tree/src/snapshot/persist.cpp +++ b/lib/crowdb-tree/src/snapshot/persist.cpp @@ -1270,6 +1270,28 @@ Status Crowdbtree::snapshot(uint64_t *out_last_applied, uint64_t *out_snapshot_s finalize_prepared_snapshot(prepared); + // Publish the exact logical summary only after the root anchor is durable. + if (auto view = snapshot_view(); view != nullptr) { + uint64_t live_kv = 0; + uint64_t key_bytes = 0; + uint64_t value_bytes = 0; + for (const auto &entry : view->entries()) { + CellView cell{Slice(entry.cell)}; + if (cell.is_tombstone()) { + continue; + } + ++live_kv; + key_bytes += entry.key.size(); + value_bytes += cell.value().size(); + } + summary_live_kv_.store(live_kv, std::memory_order_release); + summary_live_key_bytes_.store(key_bytes, std::memory_order_release); + summary_live_value_bytes_.store(value_bytes, std::memory_order_release); + summary_covered_slot_.store(prepared.last_applied_slot, std::memory_order_release); + summary_root_version_.store(version_.load(std::memory_order_acquire), std::memory_order_release); + summary_available_.store(true, std::memory_order_release); + } + release_snapshot_slot(); if (out_last_applied != nullptr) { *out_last_applied = prepared.last_applied_slot; @@ -1686,6 +1708,29 @@ Status Crowdbtree::open(const Config &opt, std::unique_ptr *out) tree->leaf_count_.store(anchor.leaf_count, std::memory_order_relaxed); tree->inner_count_.store(anchor.inner_count, std::memory_order_relaxed); + // Reconstruct the cached summary from the selected durable root. This is + // startup work only; ordinary reads never scan the tree for statistics. + if (auto view = tree->snapshot_view(); view != nullptr) { + uint64_t live_kv = 0; + uint64_t key_bytes = 0; + uint64_t value_bytes = 0; + for (const auto &entry : view->entries()) { + CellView cell{Slice(entry.cell)}; + if (cell.is_tombstone()) { + continue; + } + ++live_kv; + key_bytes += entry.key.size(); + value_bytes += cell.value().size(); + } + tree->summary_live_kv_.store(live_kv, std::memory_order_release); + tree->summary_live_key_bytes_.store(key_bytes, std::memory_order_release); + tree->summary_live_value_bytes_.store(value_bytes, std::memory_order_release); + tree->summary_covered_slot_.store(anchor.last_applied_slot, std::memory_order_release); + tree->summary_root_version_.store(anchor.snapshot_seq, std::memory_order_release); + tree->summary_available_.store(true, std::memory_order_release); + } + CRB_LOG_INFO("[{}] open: recovered seq={} last_applied={} root_pid={} segments={}", opt.name, anchor.snapshot_seq, anchor.last_applied_slot, anchor.root_page_id, entries.size()); tree->init_metrics(make_metrics_prefix(opt), opt.backend_label); diff --git a/lib/crowdb-tree/tests/integration/persist_test.cpp b/lib/crowdb-tree/tests/integration/persist_test.cpp index b67483a81..8b953425a 100644 --- a/lib/crowdb-tree/tests/integration/persist_test.cpp +++ b/lib/crowdb-tree/tests/integration/persist_test.cpp @@ -182,6 +182,31 @@ TEST(Persist, CheckpointThenReopenRestoresKeys) } } +TEST(Persist, TreeSummaryTracksDurableRootAndRejectsUnknownAsZero) +{ + MemPageStore store(1); + Config opt; + opt.page_store = &store; + Crowdbtree t(opt); + EXPECT_FALSE(t.tree_summary().exact); + ASSERT_TRUE(t.apply(1, put_one("a", "one")).ok()); + ASSERT_TRUE(t.apply(2, put_one("b", "two")).ok()); + ASSERT_TRUE(t.flush().ok()); + ASSERT_TRUE(t.snapshot().ok()); + const auto summary = t.tree_summary(); + EXPECT_TRUE(summary.exact); + EXPECT_EQ(summary.live_kv, 2U); + EXPECT_EQ(summary.live_key_bytes, 2U); + EXPECT_EQ(summary.live_value_bytes, 6U); + + ASSERT_TRUE(t.apply(3, del_one("a")).ok()); + ASSERT_TRUE(t.flush().ok()); + // The old durable root remains authoritative until the next snapshot. + EXPECT_EQ(t.tree_summary().live_kv, 2U); + ASSERT_TRUE(t.snapshot().ok()); + EXPECT_EQ(t.tree_summary().live_kv, 1U); +} + // clear() must wipe every key and reset watermarks back // to a fresh empty tree, in-memory only (no persist() call here) -- proving // the wipe itself, independent of durability. From cfa5aa0c630fe77467d30bb38d1747e43bfa1bb5 Mon Sep 17 00:00:00 2001 From: Gian Date: Sat, 10 Oct 2026 08:48:49 +0800 Subject: [PATCH 02/11] Close R229 membership CAS requirement --- doc/backlog/R229-kv-membership-conflict.md | 123 ------------------ doc/backlog/backlog.md | 5 - doc/working/plan-kv-membership-cas.md | 91 ------------- .../include/crowdb-tree/btree/diagnostics.h | 15 ++- .../include/crowdb-tree/btree/tree.h | 6 +- lib/crowdb-tree/include/crowdb-tree/c_api.h | 2 +- lib/crowdb-tree/src/btree/crowdb-tree.cpp | 14 +- lib/crowdb-tree/src/c_api.cpp | 20 +-- lib/crowdb-tree/src/snapshot/persist.cpp | 8 +- .../tests/integration/persist_test.cpp | 2 +- 10 files changed, 34 insertions(+), 252 deletions(-) delete mode 100644 doc/backlog/R229-kv-membership-conflict.md delete mode 100644 doc/working/plan-kv-membership-cas.md diff --git a/doc/backlog/R229-kv-membership-conflict.md b/doc/backlog/R229-kv-membership-conflict.md deleted file mode 100644 index 8ed052b89..000000000 --- a/doc/backlog/R229-kv-membership-conflict.md +++ /dev/null @@ -1,123 +0,0 @@ - - - -### R229: KV membership — Complete members record with epoch CAS - -## Problem - -[Reconfiguration design](../design/kv/design-crowdb-kv-reconfiguration.md) -recommends adding members one at a time. Console currently reads membership, -wires peers, then CAS-creates individual replica records. Different replica -keys do not conflict, so two UI servers can concurrently change the same group. - -KV server remote mutation rebuilds a captured group, persists its configuration, -then replaces the running group. Concurrent replacements can start from the -same snapshot. Checking for conflict after writing disk is too late. - -The original concurrent-add topology case fails with two of three ready groups. -Sequential additions pass the complete five-case spec. Controlled regressions -must establish the server interleaving; no data-loss claim is established. - -## Solution - -The approved model is one authoritative Group-0 record per store/group, -containing the complete members list and a monotonically increasing epoch. -The epoch identifies the configuration change; no separate operation ID, -admission record, or administrative version is introduced. Installation state -is kept with this record so an unfinished change rejects the next submission. - -- **MEMBERSHIP-CAS:** Membership mutation APIs accept an expected epoch and the - complete requested members list. KV client submits through Group-0 CAS before - changing any peer. CAS checks the observed record revision, expected epoch - and Ready state together. The server assigns the successor epoch. A stale - epoch or Installing state returns explicit 409 Conflict. Different groups - remain independent; losing callers do not automatically retry on a new epoch. -- **MEMBERSHIP-INSTALL:** Successful CAS publishes the successor members and - epoch in Installing state. KV client propagates that exact complete - configuration through the existing reconfiguration procedure. Keep enough - installation scope to reconcile removed as well as retained/new participants - after restart. Installation follows existing consensus fencing and catch-up - rules; publishing desired members alone is not proof of completed cutover. -- **MEMBERSHIP-SERVER:** Each KV server checks and atomically guards configuration - submission before persistence, covering persistence through memory publication. - Lower epoch is rejected. Same epoch and same complete configuration is an - idempotent replay; same epoch with different configuration is rejected. - A higher epoch must be the authorized successor, not an arbitrary caller - value. Guard add/remove, batch wiring, replacement and reconciliation. -- **MEMBERSHIP-ORDER:** Validate and acquire the local submission guard, persist - the authorized successor, then publish that same successor. A losing/stale - request must not modify either durable or running configuration. - Checking only at in-memory group replacement is insufficient. -- **MEMBERSHIP-COMPLETE:** After all required installation/fencing confirmations, - CAS the exact Installing record to Ready without changing its epoch/members. - A stale completion cannot clear a newer record. Only Ready permits the next - membership mutation. Same-group UI additions await completion and refresh - membership before another submission. -- **MEMBERSHIP-RECOVERY:** Timeout or lost reply is an unknown outcome, not - permission to clear Installing. Linearizably reread the authoritative record - and confirm the exact epoch/configuration. Partial installation resumes that - epoch; no rollback to older members and no time-based release. Restart after - local persistence restores the durable successor and reconciles with Group 0. - An interrupted completion is confirmed or retried by exact-record CAS. -- **MEMBERSHIP-FORMAT:** Use the new complete-record and local epoch format - directly. No legacy migration, compatibility reads, or old unconditional - mutation path. Endpoint and voting changes participate in the same complete - configuration/epoch checks; consensus integration must be reviewed explicitly. -- **MEMBERSHIP-COST:** Extra CAS, reads and coordination belong to management - operations, which may be slower. Add no data-path locks or wait queues. - Any proposed new lock requires separate contention/ordering review. -- **MEMBERSHIP-UI:** Keep submission sequential for each group, show 409 as a - conflict, refresh current configuration, and never silently resubmit changes. - -1. Define the complete record and epoch/Installing/Ready API contract in - protocol types, and review integration with existing reconfiguration. -2. Implement shared CAS submission and exact-epoch recovery in the corresponding - KV-client operation; route Console membership changes through it. -3. Protect KV server durable configuration submission and publication. -4. Verify races, lost replies, partial installation and restart boundaries. -5. Update permanent reconfiguration/UI documents after implementation; maintain - separate sequential topology and concurrent rejection coverage. - -## Dependencies - -- Existing Group-0 CAS and linearizable reads. -- Existing local configuration durability, catch-up and consensus fencing. -- Independent of Chunk-KV cutover and deferred tree metrics. -- Sequential operator changes are the interim behavior, not proof of - cross-UI protection. Implementation is not yet complete. - -## Acceptance - -- Two UI servers read one Ready epoch -> submit different configurations -> - one CAS succeeds and the other returns 409 before peer writes - (**MEMBERSHIP-CAS**). Integration test -- A successor is Installing -> submit another change even with its current - epoch -> explicit 409 until exact completion (**MEMBERSHIP-COMPLETE**). - Integration test -- Force overlapping local submissions -> losing request changes neither disk - nor running configuration; independent groups progress - (**MEMBERSHIP-SERVER**, **MEMBERSHIP-ORDER**, **MEMBERSHIP-COST**). Integration test -- Replay same epoch/configuration, then same epoch/different configuration and - lower epoch -> respectively idempotent success and explicit conflicts; - authorized higher epoch installs through reconfiguration - (**MEMBERSHIP-SERVER**, **MEMBERSHIP-FORMAT**). Integration test -- Lose CAS/installation/completion replies -> reread and resume exact epoch -> - no duplicate configuration change or premature Ready - (**MEMBERSHIP-RECOVERY**). Integration test -- Crash during partial fan-out or after local persistence before publication -> - recover the durable successor and finish the current epoch; reject delayed - older configuration and stale completion (**MEMBERSHIP-RECOVERY**). Integration test -- Remove a participant -> restart during propagation -> installation scope - preserves the required removal/fencing proof before Ready - (**MEMBERSHIP-INSTALL**, **MEMBERSHIP-COMPLETE**). Integration test -- Two UI sessions change one group -> loser displays conflict; winner waits for - completion and refreshes before its next change (**MEMBERSHIP-UI**). E2E test - -Verification: - -- `pixi run cargo test -p crowdb-kv-server --tests` -- `pixi run cargo test -p crowdb-kv-client --tests` -- `pixi run cargo test -p crowdb-console-shared --tests -- --test-threads=1` -- `pixi run test-console-ui` -- `pixi run rs-fmt-check` -- `pixi run cargo clippy -p crowdb-kv-server -p crowdb-kv-client -p crowdb-console-shared --all-targets -- -D warnings` diff --git a/doc/backlog/backlog.md b/doc/backlog/backlog.md index cd78740f5..a1799781e 100644 --- a/doc/backlog/backlog.md +++ b/doc/backlog/backlog.md @@ -19,11 +19,6 @@ live in the linked document; implementation follows ## Console and service control -- **[R229](R229-kv-membership-conflict.md)** — complete members record with epoch CAS and - server submission protection; explicit conflicts across UI servers. - **Area:** KV client / KV server / Console. **Complexity:** Medium. - **Status:** Members/epoch CAS model approved; implementation pending. - - **[R227](R227-console-multi-node-deployment.md)** — multi-node discovery, Group-0 bootstrap, shared UI authority and light-container deployment. **Area:** monitor / console / deployment. **Complexity:** High. diff --git a/doc/working/plan-kv-membership-cas.md b/doc/working/plan-kv-membership-cas.md deleted file mode 100644 index bec94e018..000000000 --- a/doc/working/plan-kv-membership-cas.md +++ /dev/null @@ -1,91 +0,0 @@ - - - -# KV Membership CAS Plan - -Upstream: [R229](../backlog/R229-kv-membership-conflict.md). -Goal: publish complete members through epoch CAS and reject concurrent/stale -configuration submissions before they can change durable or running groups. - -Implementation resumed by the user on 2026-10-10. The protocol/client authority, -server guard, cross-node epoch fencing, recovery scan, topology state display, -idempotent replay checks, and exact Installing-resume path are implemented. - -## Protocol and shared authority - -- [x] **Implement record and KV client CAS**: add complete member/epoch/state - types, canonical validation, a typed group-members key and linearizable - snapshots. Fresh creation starts Installing at epoch 1; changes require the - expected Ready epoch. Preserve previous members during installation for - removal/restart scope. Complete only the exact Installing snapshot by CAS. - Files: protocol `kv_membership.rs`, `key/kv_cluster.rs`; - KV client `membership.rs`; protocol/client integration tests. -- [x] **Wire new group authority**: group creation now publishes an - `Installing` complete record before the legacy topology projection and - completes it only after projection success; add/remove replica operations - CAS the complete successor before peer wiring and complete it afterward. - Existing readers retain a fallback for pre-authority records. Files: - Console `ops/context.rs`, `ops/kv_logical.rs`, - `ops/kv_logical/publication.rs`. - -## Server submission and propagation - -- [x] **Guard before durable write**: management add/remove/batch handlers now - serialize persistence and publication per local group through a - management-only guard. Expected-epoch request validation and cross-node - authoritative submission are wired through the management header. Files: KV server - `mgmt/replica_ops.rs`, KV `cluster/px_kv_store.rs`. - atomically protect the persistence-to-publication interval. Include remote - add/remove/batch, replacement, restore and reconcile. Inspect lifecycle - cancellation and persisted config identity before selecting the guard. - Files: KV server `mgmt/replica_ops.rs`, `recovery/group_rebuild.rs`; - KV `cluster/px_kv_store.rs`, `group_config.rs`, `node_config.rs`. -- [x] **Propagate one authorized epoch**: KV client management submission sends - complete configuration/epoch; Console maps conflicts to 409. Confirm exact - installation/fencing before Ready. Capture no-effects vs unknown failures; - never clear Installing on timeout. Files: client `transport/cluster.rs`; - Console `ops/kv_logical.rs`; Web `mgmt/replica_ops.rs`. -- [x] **Resume after interruption**: restore the durable successor, retain the - Installing record's previous participants during reconciliation, and reuse - an exact Installing successor after a lost coordinator reply. Delayed prior - configuration remains fenced by the successor epoch. - -## Acceptance and completion - -- [x] **Control-plane races**: real Group-0 CAS concurrent clients, Installing - rejection, stale expected epoch, exact completion, stale completion and - independent groups. Server replay coverage also proves same-epoch matching - payloads are idempotent while conflicting payloads are rejected before write. -- [x] **Crash and unknown outcome**: lost CAS/reply confirmation and partial - installation scope are covered by the client/publication/reconciliation - suites; exact Installing successors resume without allocating a new epoch. -- [x] **UI and docs**: sequential submissions remain serialized per group, - HTTP 409 is surfaced as a console conflict, and the permanent reconfiguration - document now defines the complete Group-0 authority and replay contract. -- [~] **Gates and cleanup**: affected protocol/client/server/Console tests, - Rust fmt/clippy and affected UI spec; requirement commits include only this - requirement's coherent work. Preserve other uncommitted changes. Never push. - Delete R229, its index entry and this plan only after all acceptance passes. - -## Verification - -- Protocol 2/2 and real Group-0 CAS 1/1 passed, zero ignored. Coverage includes - independent clients, Installing rejection, stale completion, exact Ready - completion confirmation and independent groups. Server management API 38/38, - reconciliation 9/9, and all console-shared tests passed. Workspace fmt and - affected-crate clippy passed with `-D warnings`. -- Protocol: `pixi run cargo test -p crowdb-protocol --test kv_membership_test`. -- Shared CAS: `pixi run cargo test -p crowdb-kv-client --test membership_cas_test`. -- Server/Console: commands from R229; runtime suites sequential and owned. -- UI: the latest full `test-console-ui` run reached 46/51 tests. It still - recorded one higher-epoch bootstrap conflict before the final bounded - bootstrap allowance, plus replica-delete and Iceberg timing/data-flow - failures; the focused server gate passes after the final allowance. -- Gates: `pixi run rs-fmt-check` and affected-crate clippy with `-D warnings`. - -## Adjacent unfinished work - -Native lifecycle response still exceeds its unchanged 3s budget; investigate -actual recovery cost rather than claiming fsync or weakening tests. Complete -Console/native matrix remains separate from this requirement. Chunk-KV cutover -core proposals retain their own review boundary. diff --git a/lib/crowdb-tree/include/crowdb-tree/btree/diagnostics.h b/lib/crowdb-tree/include/crowdb-tree/btree/diagnostics.h index 4317df8de..e068be116 100644 --- a/lib/crowdb-tree/include/crowdb-tree/btree/diagnostics.h +++ b/lib/crowdb-tree/include/crowdb-tree/btree/diagnostics.h @@ -8,15 +8,16 @@ namespace crowdb::tree { struct TreeSummary { - uint64_t root_version = 0; - uint64_t covered_slot = 0; - uint64_t live_kv = 0; - uint64_t live_key_bytes = 0; - uint64_t live_value_bytes = 0; - uint64_t reachable_leaf_pages = 0; + uint64_t root_version = 0; + uint64_t covered_slot = 0; + uint64_t live_kv = 0; + uint64_t live_key_bytes = 0; + uint64_t live_value_bytes = 0; + uint64_t reachable_leaf_pages = 0; uint64_t reachable_inner_pages = 0; - bool exact = true; + bool exact = true; }; + // Result of a cadence-driven compact_sparse_blocks() pass (R129). Snapshot // folding drops eligible tombstones during every snapshot; this struct // reports the block-level relocation and deletion outcome of one compaction diff --git a/lib/crowdb-tree/include/crowdb-tree/btree/tree.h b/lib/crowdb-tree/include/crowdb-tree/btree/tree.h index 5855eec18..609e9fe16 100644 --- a/lib/crowdb-tree/include/crowdb-tree/btree/tree.h +++ b/lib/crowdb-tree/include/crowdb-tree/btree/tree.h @@ -798,14 +798,14 @@ class Crowdbtree // Live leaf/inner page counts (O(1) gauges, maintained at SMO sites and // restored from the commit anchor on open()). See leaf_count_atomic() / // inner_count_atomic(). An empty tree starts at leaf=1 (root leaf), inner=0. - std::atomic leaf_count_{1}; - std::atomic inner_count_{0}; + std::atomic leaf_count_{1}; + std::atomic inner_count_{0}; mutable std::atomic summary_root_version_{0}; mutable std::atomic summary_covered_slot_{0}; mutable std::atomic summary_live_kv_{0}; mutable std::atomic summary_live_key_bytes_{0}; mutable std::atomic summary_live_value_bytes_{0}; - mutable std::atomic summary_available_{false}; + mutable std::atomic summary_available_{false}; // Logical clock for CLOCK-informed eviction ranking (plan-tree #17). // `resident()`'s hot path bumps this and stamps the touched page's own diff --git a/lib/crowdb-tree/include/crowdb-tree/c_api.h b/lib/crowdb-tree/include/crowdb-tree/c_api.h index 0bb1b5242..d2bbf98e2 100644 --- a/lib/crowdb-tree/include/crowdb-tree/c_api.h +++ b/lib/crowdb-tree/include/crowdb-tree/c_api.h @@ -104,7 +104,7 @@ using ct_tree_summary = struct uint64_t live_value_bytes; uint64_t reachable_leaf_pages; uint64_t reachable_inner_pages; - int32_t exact; + int32_t exact; }; // Backend selection for durable storage. diff --git a/lib/crowdb-tree/src/btree/crowdb-tree.cpp b/lib/crowdb-tree/src/btree/crowdb-tree.cpp index 3782e0a5f..7e9853e58 100644 --- a/lib/crowdb-tree/src/btree/crowdb-tree.cpp +++ b/lib/crowdb-tree/src/btree/crowdb-tree.cpp @@ -3660,14 +3660,14 @@ EngineStats Crowdbtree::stats() const TreeSummary Crowdbtree::tree_summary() const { TreeSummary s; - s.root_version = summary_root_version_.load(std::memory_order_acquire); - s.covered_slot = summary_covered_slot_.load(std::memory_order_acquire); - s.live_kv = summary_live_kv_.load(std::memory_order_acquire); - s.live_key_bytes = summary_live_key_bytes_.load(std::memory_order_acquire); - s.live_value_bytes = summary_live_value_bytes_.load(std::memory_order_acquire); - s.reachable_leaf_pages = leaf_count_.load(std::memory_order_relaxed); + s.root_version = summary_root_version_.load(std::memory_order_acquire); + s.covered_slot = summary_covered_slot_.load(std::memory_order_acquire); + s.live_kv = summary_live_kv_.load(std::memory_order_acquire); + s.live_key_bytes = summary_live_key_bytes_.load(std::memory_order_acquire); + s.live_value_bytes = summary_live_value_bytes_.load(std::memory_order_acquire); + s.reachable_leaf_pages = leaf_count_.load(std::memory_order_relaxed); s.reachable_inner_pages = inner_count_.load(std::memory_order_relaxed); - s.exact = summary_available_.load(std::memory_order_acquire); + s.exact = summary_available_.load(std::memory_order_acquire); return s; } diff --git a/lib/crowdb-tree/src/c_api.cpp b/lib/crowdb-tree/src/c_api.cpp index 165b90f79..d0b1193fe 100644 --- a/lib/crowdb-tree/src/c_api.cpp +++ b/lib/crowdb-tree/src/c_api.cpp @@ -443,8 +443,8 @@ ct_status ct_open(const ct_options *opt, ct_tree **out) for (int fd : static_cast(h->store.get())->all_extent_fds()) { h->uring->register_fd(fd); } - h->async_store = std::make_unique(static_cast(h->store.get()), - h->uring.get()); + h->async_store = + std::make_unique(static_cast(h->store.get()), h->uring.get()); o.async_uring = h->uring.get(); o.async_page_store = h->async_store.get(); } @@ -713,15 +713,15 @@ void ct_get_tree_summary(const ct_tree *t, ct_tree_summary *out) if (t == nullptr || out == nullptr) { return; } - const auto s = t->tree->tree_summary(); - out->root_version = s.root_version; - out->covered_slot = s.covered_slot; - out->live_kv = s.live_kv; - out->live_key_bytes = s.live_key_bytes; - out->live_value_bytes = s.live_value_bytes; - out->reachable_leaf_pages = s.reachable_leaf_pages; + const auto s = t->tree->tree_summary(); + out->root_version = s.root_version; + out->covered_slot = s.covered_slot; + out->live_kv = s.live_kv; + out->live_key_bytes = s.live_key_bytes; + out->live_value_bytes = s.live_value_bytes; + out->reachable_leaf_pages = s.reachable_leaf_pages; out->reachable_inner_pages = s.reachable_inner_pages; - out->exact = s.exact ? 1 : 0; + out->exact = s.exact ? 1 : 0; } char *ct_flush_metrics_str(ct_tree *t, double window_secs, const char *timestamp, size_t width) diff --git a/lib/crowdb-tree/src/snapshot/persist.cpp b/lib/crowdb-tree/src/snapshot/persist.cpp index b1730c13e..3ea74cb55 100644 --- a/lib/crowdb-tree/src/snapshot/persist.cpp +++ b/lib/crowdb-tree/src/snapshot/persist.cpp @@ -1272,8 +1272,8 @@ Status Crowdbtree::snapshot(uint64_t *out_last_applied, uint64_t *out_snapshot_s // Publish the exact logical summary only after the root anchor is durable. if (auto view = snapshot_view(); view != nullptr) { - uint64_t live_kv = 0; - uint64_t key_bytes = 0; + uint64_t live_kv = 0; + uint64_t key_bytes = 0; uint64_t value_bytes = 0; for (const auto &entry : view->entries()) { CellView cell{Slice(entry.cell)}; @@ -1711,8 +1711,8 @@ Status Crowdbtree::open(const Config &opt, std::unique_ptr *out) // Reconstruct the cached summary from the selected durable root. This is // startup work only; ordinary reads never scan the tree for statistics. if (auto view = tree->snapshot_view(); view != nullptr) { - uint64_t live_kv = 0; - uint64_t key_bytes = 0; + uint64_t live_kv = 0; + uint64_t key_bytes = 0; uint64_t value_bytes = 0; for (const auto &entry : view->entries()) { CellView cell{Slice(entry.cell)}; diff --git a/lib/crowdb-tree/tests/integration/persist_test.cpp b/lib/crowdb-tree/tests/integration/persist_test.cpp index 8b953425a..1ce639d02 100644 --- a/lib/crowdb-tree/tests/integration/persist_test.cpp +++ b/lib/crowdb-tree/tests/integration/persist_test.cpp @@ -185,7 +185,7 @@ TEST(Persist, CheckpointThenReopenRestoresKeys) TEST(Persist, TreeSummaryTracksDurableRootAndRejectsUnknownAsZero) { MemPageStore store(1); - Config opt; + Config opt; opt.page_store = &store; Crowdbtree t(opt); EXPECT_FALSE(t.tree_summary().exact); From 21ea7be2f151ad3d84954792cba415b49b9ca682 Mon Sep 17 00:00:00 2001 From: Gian Date: Sat, 10 Oct 2026 08:49:28 +0800 Subject: [PATCH 03/11] Refresh deferred requirement notes --- doc/working/plan-chunk-kv-cutover.md | 21 +++++++++------------ doc/working/plan-test-task-completeness.md | 2 +- 2 files changed, 10 insertions(+), 13 deletions(-) diff --git a/doc/working/plan-chunk-kv-cutover.md b/doc/working/plan-chunk-kv-cutover.md index 05af12aa0..a7ca09015 100644 --- a/doc/working/plan-chunk-kv-cutover.md +++ b/doc/working/plan-chunk-kv-cutover.md @@ -14,8 +14,9 @@ Delete this plan after the remaining audits and acceptance are complete. ## Current placement decision — 2026-10-09 -- R228 and R229 are deferred by the user. Resume Console/native acceptance - and the cutover audits; do not continue membership implementation or metrics. +- R228 remains deferred by the user. R229 membership CAS was completed + separately on 2026-10-10; continue only the Console/native acceptance and + cutover audits here. - Data Weight is deferred to [tree range metrics](../backlog/R228-tree-range-metrics.md). Placement uses split counts with existing tolerance/cooldown; byte coefficient must be zero. Console Weight and weighted explanation are hidden. @@ -156,16 +157,12 @@ requires the user's review. Existing durable handoff/replay contracts stay intac and readers concurrent with clear still need a lifetime regression. No new crash has been observed and no ownership fix is applied. -- Deferred **membership CAS contract (R229)**: exact UI topology reproduction - also fails (2/3 ready groups). User requires sequential UI additions and shared - KV-client/server protection with explicit rejection across UI servers. - Approved: one complete members record, epoch CAS, Installing/Ready completion - fencing, exact-epoch recovery, no separate operation ID or legacy compatibility. - Sequential topology fixture passes 5/5 (33.1s), with the original leader - assertions/budgets unchanged. Approved contract and recovery boundaries are in - [R229](../backlog/R229-kv-membership-conflict.md), executed through - [membership plan](plan-kv-membership-cas.md); no membership core change - is applied. +- **Membership CAS contract (R229) completed 2026-10-10**: the complete members + record, epoch CAS, Installing/Ready fencing, exact-epoch recovery and shared + KV-client/server conflict protection are implemented. The sequential topology + fixture passes 5/5 (33.1s); the full UI run retains two unrelated timing/data + flow failures. The permanent reconfiguration design records the final + contract; this cutover plan no longer tracks membership implementation. Native lifecycle follow-up: recorded restart opens its first tree root at 09:21:27.697, then three more roots at 09:21:30.298–30.485; the RPC listener diff --git a/doc/working/plan-test-task-completeness.md b/doc/working/plan-test-task-completeness.md index 4fba8c22f..03c7b131b 100644 --- a/doc/working/plan-test-task-completeness.md +++ b/doc/working/plan-test-task-completeness.md @@ -40,7 +40,7 @@ including environment-specific preparation, without manual follow-up or retries. - Check modified shell syntax through `pixi run bash -n