This is an automated email from the ASF dual-hosted git repository. numinnex pushed a commit to branch remove_namespace_field in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 91ac1864feb0c4afb2a70c89d90d1bdb5a6ce573 Merge: 9c6e71ffd 50518c448 Author: Grzegorz Koszyk <[email protected]> AuthorDate: Fri Aug 7 14:43:09 2026 +0200 merge master .github/actions/cpp-bazel/pre-merge/action.yml | 8 +- .github/actions/php/pre-merge/action.yml | 1 + .../actions/python-maturin/pre-merge/action.yml | 3 + .github/workflows/coverage-baseline.yml | 69 +- Cargo.lock | 54 +- Cargo.toml | 9 +- bdd/python/uv.lock | 6 +- core/ai/mcp/src/service/mod.rs | 5 +- core/bench/Cargo.toml | 2 +- core/bench/src/main.rs | 14 + core/binary_protocol/Cargo.toml | 2 +- core/binary_protocol/src/consensus/header.rs | 58 +- core/binary_protocol/src/consensus/operation.rs | 15 + core/binary_protocol/src/responses/messages/mod.rs | 2 + .../src/responses/messages/send_messages.rs | 284 ++++ core/cli/Cargo.toml | 2 +- core/common/Cargo.toml | 2 +- core/common/src/error/iggy_error.rs | 5 +- core/common/src/http/messages/send_messages.rs | 58 +- core/common/src/lib.rs | 4 + core/common/src/traits/binary_impls/messages.rs | 137 +- core/common/src/traits/binary_impls/mod.rs | 2 + core/common/src/traits/message_client.rs | 15 +- core/common/src/traits/mod.rs | 2 + core/common/src/types/message/mod.rs | 4 +- core/common/src/types/partition/mod.rs | 2 +- core/configs/Cargo.toml | 1 + core/configs/src/server_ng_config/cluster.rs | 987 +++++++++++- core/configs/src/server_ng_config/defaults.rs | 6 + core/configs/src/server_ng_config/displays.rs | 20 +- core/configs/src/server_ng_config/partition.rs | 76 + core/configs/src/server_ng_config/validators.rs | 29 + core/configs_derive/src/config_env.rs | 22 +- core/connectors/runtime/Cargo.toml | 2 +- core/connectors/sinks/clickhouse_sink/Cargo.toml | 2 +- core/connectors/sinks/delta_sink/Cargo.toml | 2 +- core/connectors/sinks/doris_sink/Cargo.toml | 2 +- .../connectors/sinks/elasticsearch_sink/Cargo.toml | 2 +- core/connectors/sinks/http_sink/Cargo.toml | 2 +- core/connectors/sinks/iceberg_sink/Cargo.toml | 2 +- core/connectors/sinks/influxdb_sink/Cargo.toml | 2 +- core/connectors/sinks/meilisearch_sink/Cargo.toml | 2 +- core/connectors/sinks/mongodb_sink/Cargo.toml | 2 +- core/connectors/sinks/postgres_sink/Cargo.toml | 2 +- core/connectors/sinks/quickwit_sink/Cargo.toml | 2 +- core/connectors/sinks/s3_sink/Cargo.toml | 2 +- core/connectors/sinks/stdout_sink/Cargo.toml | 2 +- core/connectors/sinks/surrealdb_sink/Cargo.toml | 2 +- .../sources/elasticsearch_source/Cargo.toml | 2 +- core/connectors/sources/influxdb_source/Cargo.toml | 2 +- core/connectors/sources/postgres_source/Cargo.toml | 2 +- core/connectors/sources/random_source/Cargo.toml | 2 +- core/consensus/src/impls.rs | 27 +- core/consensus/src/lib.rs | 4 +- core/consensus/src/state_transfer.rs | 59 +- core/consensus/src/vsr_state.rs | 112 +- core/integration/src/bench_utils.rs | 101 +- .../src/harness/orchestrator/harness.rs | 17 + .../tests/cluster/metadata_checkpoint_restart.rs | 138 +- .../tests/cluster/partition_state_transfer.rs | 227 ++- core/integration/tests/sdk/mcp_parity.rs | 608 ------- core/integration/tests/sdk/messages.rs | 172 -- core/integration/tests/sdk/mod.rs | 5 +- core/integration/tests/sdk/producer/background.rs | 6 +- core/integration/tests/sdk/send_confirmation.rs | 268 +++ .../tests/server/cluster_metadata_vsr.rs | 229 ++- .../tests/server/cluster_view_durability_vsr.rs | 17 +- core/integration/tests/server/general.rs | 19 +- core/integration/tests/server/http_client.rs | 19 +- core/integration/tests/server/http_tls.rs | 112 +- .../server/metadata_checkpoint_recovery_vsr.rs | 138 -- core/integration/tests/server/mod.rs | 4 - .../tests/server/partition_view_durability_vsr.rs | 90 +- .../server/scenarios/authentication_scenario.rs | 1 + .../server/scenarios/create_message_payload.rs | 148 -- core/integration/tests/server/scenarios/mod.rs | 1 - .../server/scenarios/purge_delete_scenario.rs | 95 +- core/journal/src/local_gate.rs | 109 +- core/journal/src/superblock.rs | 10 +- core/metadata/src/impls/metadata.rs | 3 +- core/metadata/src/impls/recovery.rs | 1 + core/metadata/src/lib.rs | 4 + core/partitions/Cargo.toml | 1 + core/partitions/src/iggy_partition.rs | 1172 ++++++++++++-- core/partitions/src/journal.rs | 72 + core/partitions/src/lib.rs | 2 +- core/partitions/src/offset_storage.rs | 21 +- core/partitions/src/state_transfer.rs | 1510 ++++++++++++++--- core/sdk/Cargo.toml | 3 +- .../src/client_wrappers/binary_message_client.rs | 3 +- core/sdk/src/clients/binary_message.rs | 3 +- core/sdk/src/clients/producer.rs | 167 +- core/sdk/src/clients/producer_dispatcher.rs | 7 +- core/sdk/src/clients/producer_error_callback.rs | 11 +- core/sdk/src/clients/producer_sharding.rs | 11 +- core/sdk/src/http/messages.rs | 94 +- core/sdk/src/prelude.rs | 9 +- core/server-ng/Cargo.toml | 8 +- core/server-ng/config.toml | 77 + core/server-ng/src/bootstrap.rs | 180 ++- core/server-ng/src/cluster_meta.rs | 125 +- core/server-ng/src/dispatch.rs | 72 +- core/server-ng/src/dispatch/authz.rs | 11 +- core/server-ng/src/http.rs | 37 +- core/server-ng/src/http/error.rs | 126 +- core/server-ng/src/http/extractor.rs | 22 +- core/server-ng/src/http/handlers.rs | 85 +- core/server-ng/src/http/reads.rs | 3 +- core/server-ng/src/http/reply.rs | 146 +- core/server-ng/src/http/state.rs | 22 +- core/server-ng/src/http/submit.rs | 11 +- core/server-ng/src/http/tls.rs | 15 +- core/server-ng/src/partition_helpers.rs | 131 +- core/server-ng/src/partition_reconciler.rs | 26 +- core/server-ng/src/responses.rs | 12 +- core/server-ng/src/segment_recovery.rs | 286 +++- core/server-ng/src/server_error.rs | 86 +- core/server-ng/src/session_manager.rs | 28 + core/server-ng/tests/sdk_e2e.rs | 175 +- core/server_common/src/consensus_message.rs | 32 +- core/shard/src/lib.rs | 1702 ++++++++++++++++---- core/shard/src/metrics.rs | 22 + core/shard/src/router.rs | 90 +- core/simulator/src/lib.rs | 99 +- core/simulator/src/workload/ops/mod.rs | 2 +- examples/node/package-lock.json | 26 +- examples/node/package.json | 2 +- examples/python/uv.lock | 6 +- foreign/cpp/MODULE.bazel | 2 +- foreign/cpp/MODULE.bazel.lock | 4 +- foreign/cpp/src/client.rs | 7 +- foreign/cpp/src/lib.rs | 27 +- foreign/cpp/src/type_conversion.rs | 30 +- foreign/cpp/tests/e2e/message.cpp | 9 +- foreign/node/package-lock.json | 12 +- foreign/node/package.json | 4 +- foreign/node/scripts/check-vsr-protocol.mjs | 52 +- foreign/node/src/client/client.utils.test.ts | 23 +- foreign/node/src/client/client.utils.ts | 3 - foreign/node/src/e2e/tcp.send-message.e2e.ts | 27 +- foreign/node/src/index.ts | 2 + .../src/wire/message/send-messages.command.test.ts | 110 +- .../node/src/wire/message/send-messages.command.ts | 81 +- foreign/node/src/wire/vsr/header.test.ts | 8 - foreign/node/src/wire/vsr/header.ts | 23 +- foreign/node/src/wire/vsr/index.ts | 3 - foreign/node/src/wire/vsr/namespace.test.ts | 342 ---- foreign/node/src/wire/vsr/namespace.ts | 216 --- foreign/node/src/wire/vsr/register.ts | 4 +- foreign/php/iggy-php.stubs.php | 59 +- foreign/php/src/client.rs | 9 +- foreign/php/src/lib.rs | 4 +- foreign/php/src/message_iterator.rs | 3 + foreign/php/src/send_message.rs | 87 +- foreign/php/tests/IggySdkTest.php | 20 + foreign/python/Cargo.toml | 4 +- foreign/python/apache_iggy.pyi | 289 +++- foreign/python/pyproject.toml | 6 +- foreign/python/src/client.rs | 122 +- foreign/python/src/consumer.rs | 32 +- foreign/python/src/lib.rs | 9 +- foreign/python/src/send_message.rs | 98 +- foreign/python/src/topic.rs | 276 +++- foreign/python/tests/test_message_operations.py | 40 + foreign/python/tests/test_topic.py | 235 ++- foreign/python/uv.lock | 12 +- scripts/ci/coverage-baseline-affected.sh | 96 ++ scripts/ci/license-headers.sh | 7 +- web/package-lock.json | 544 +++++-- web/package.json | 10 +- 170 files changed, 10669 insertions(+), 3721 deletions(-) diff --cc core/binary_protocol/src/consensus/header.rs index b5efe8a6c,a4661cd27..0d7995059 --- a/core/binary_protocol/src/consensus/header.rs +++ b/core/binary_protocol/src/consensus/header.rs @@@ -1552,9 -1413,42 +1552,42 @@@ pub struct StateTransferTargetHeader /// Serving primary's applied frontier (`commit_min`) when the descriptor /// was built. The receiver's tail repair targets past this. pub commit_op: u64, - pub namespace: u64, + pub group: u64, pub available: u8, - pub reserved: [u8; 95], + /// Set on an `available == 0` refusal that means "not right now" rather than + /// "this node is broken". + /// + /// PARTITION arm only: it is the only side with a consecutive-failure count + /// to charge. The requester then re-arms on a flat interval instead of + /// charging that count, whose exponential backoff climbs to 1024x the retry + /// interval and is reset only by a completed install. A serving primary + /// momentarily behind its own frontier is the common case under produce + /// load. + /// + /// This and `commit_max` below claim the HEAD of what used to be the + /// reserved tail, so every pre-existing field keeps its published offset. + /// Layout compatibility only: the size assert cannot catch an equal-size + /// reshuffle, so a mid-struct insertion would silently move every field + /// after it. It says nothing about the semantics of these two -- an older + /// peer presents zeros here and serves no partition transfers at all. + pub unavailable_transient: u8, + /// Explicit padding so `commit_max` sits 8-aligned without the implicit + /// padding `NoUninit` forbids. + pub reserved_alignment: [u8; 6], + /// Serving replica's `commit_max` when the descriptor was built. + /// + /// Read by the PARTITION receiver only; the metadata arm branches on + /// `available` and falls back to journal repair without a refusal. + /// + /// A partition receiver refuses an offer from a replica that knows LESS + /// than it does: + /// without this the descriptor carried no proof of the sender's own + /// progress, and a phantom view-0 primary (a group whose directory vanished + /// boots `init()` rather than `init_as_backup()`, comes up Normal at view 0, + /// and an empty log is trivially caught up) could hand a data-holding + /// rejoiner an empty offer that unlinks its chain. + pub commit_max: u64, + pub reserved: [u8; 80], } const _: () = { assert!(size_of::<StateTransferTargetHeader>() == HEADER_SIZE); @@@ -1562,7 -1456,14 +1595,14 @@@ offset_of!(StateTransferTargetHeader, nonce) == offset_of!(StateTransferTargetHeader, reserved_frame) + size_of::<[u8; 66]>() ); - assert!(offset_of!(StateTransferTargetHeader, reserved) + size_of::<[u8; 95]>() == HEADER_SIZE); + // The pre-existing published offsets. New fields grow into the reserved + // tail only; a change that moves one of these is a wire break. + assert!(offset_of!(StateTransferTargetHeader, commit_op) == 144); - assert!(offset_of!(StateTransferTargetHeader, namespace) == 152); ++ assert!(offset_of!(StateTransferTargetHeader, group) == 152); + assert!(offset_of!(StateTransferTargetHeader, available) == 160); + assert!(offset_of!(StateTransferTargetHeader, unavailable_transient) == 161); + assert!(offset_of!(StateTransferTargetHeader, commit_max) == 168); + assert!(offset_of!(StateTransferTargetHeader, reserved) + size_of::<[u8; 80]>() == HEADER_SIZE); }; impl ConsensusHeader for StateTransferTargetHeader { diff --cc core/consensus/src/impls.rs index c8379099b,d23378f15..d2c3247cd --- a/core/consensus/src/impls.rs +++ b/core/consensus/src/impls.rs @@@ -2078,7 -2080,7 +2081,7 @@@ impl<B: MessageBus, P: Pipeline<Entry /// that will be the primary in the new view." /// /// # Panics -- /// If `header.namespace` does not match this replica's namespace. ++ /// If `header.group` does not match this replica's namespace. pub fn handle_start_view_change( &self, plane: PlaneKind, @@@ -2231,7 -2236,7 +2234,7 @@@ /// and selects as the new log the one contained in the message with the largest v'..." /// /// # Panics -- /// If `header.namespace` does not match this replica's namespace. ++ /// If `header.group` does not match this replica's namespace. pub fn handle_do_view_change( &self, plane: PlaneKind, @@@ -2519,7 -2527,7 +2522,7 @@@ /// their status to normal, and send `PrepareOK` for any uncommitted ops." /// /// # Panics -- /// If `header.namespace` does not match this replica's namespace. ++ /// If `header.group` does not match this replica's namespace. /// # Client-table maintenance /// /// Backups maintain the client-table during normal operation via @@@ -2692,9 -2700,12 +2695,9 @@@ /// to prevent old/replayed messages from suppressing view changes. /// /// # Panics -- /// If `header.namespace` does not match this replica's namespace. ++ /// If `header.group` does not match this replica's namespace. pub fn handle_commit(&self, header: &iggy_binary_protocol::CommitHeader) -> CommitOutcome { - assert_eq!( - header.namespace, self.namespace, - "Commit routed to wrong group" - ); + assert_eq!(header.group, self.group, "Commit routed to wrong group"); if self.is_primary() { // A heartbeat from the primary of an OLDER view means that @@@ -3071,12 -3086,12 +3078,12 @@@ wher op, timestamp, operation: old.operation, -- // The GROUP's namespace, never the request's: a client - // RoutedRequestHeader carries namespace 0, and journaling that - // RequestHeader carries namespace 0, and journaling that -- // would make the stored prepare route to the wrong plane -- // when repair later ships it verbatim (live replication -- // masked this; repair replay is what broke). - namespace: consensus.namespace, ++ // The GROUP's own id, never the request's: a routed request ++ // header can carry group 0, and journaling that would make ++ // the stored prepare route to the wrong plane when repair ++ // later ships it verbatim (live replication masked this; ++ // repair replay is what broke). + group: consensus.group, checksum_body, // Copied verbatim: carries the stamped acting user for client // ops (and the authenticated user on Register), so the in-apply diff --cc core/partitions/src/iggy_partition.rs index c04197605,dad3f70e3..4993c9bfc --- a/core/partitions/src/iggy_partition.rs +++ b/core/partitions/src/iggy_partition.rs @@@ -44,10 -44,13 +44,13 @@@ use iggy_binary_protocol::requests::con DeleteConsumerOffset2Request, DeleteConsumerOffsetRequest, StoreConsumerOffset2Request, StoreConsumerOffsetRequest, }; + use iggy_binary_protocol::responses::messages::{ + SendMessagesConfirmationResponse, SendMessagesResponse, + }; use iggy_binary_protocol::{ - AckLevel, GenericHeader, Operation, PrepareHeader, WireDecode, WireIdentifier, + AckLevel, GenericHeader, Operation, PrepareHeader, WireDecode, WireEncode, WireIdentifier, }; -use iggy_binary_protocol::{PrepareOkHeader, RequestHeader}; +use iggy_binary_protocol::{PrepareOkHeader, RoutedRequestHeader}; use iggy_common::{ ConsumerGroupId, ConsumerGroupOffsets, ConsumerKind, ConsumerOffset, ConsumerOffsets, IggyByteSize, IggyError, IggyExpiry, IggyTimestamp, PartitionStats, PollingKind, @@@ -530,6 -659,181 +659,181 @@@ wher } } + /// Re-seed the offset counter from a recovered superblock record, taking + /// the MAX of what the record holds and what the recovered segments already + /// proved. + /// + /// The record is a lower bound, never a completeness claim: it exists + /// because three paths leave a replica whose counter would otherwise + /// restart at 0 while the group is at N (a transfer install of an all-GC'd + /// origin, a crash inside the install's swap window, and the + /// fence-and-rebuild path, which needs no crash at all). Restarting the + /// counter is not a lag -- replicas re-stamp `base_offset` from it and + /// recompute `batch_checksum` over the result, so the next replicated + /// prepare would persist different bytes here than on every peer, silently. + /// + /// Lives HERE rather than in the server crate so the boot paths and the + /// simulator share one implementation. A copy in the harness was a copy of + /// the max rule that had lost the max, in the one place built to catch + /// violations of it. + pub fn restore_offset_frontier(&mut self, recovered: Option<&consensus::VsrState>) { + let Some(frontier) = recovered + .map(|state| state.offset_frontier) + .filter(|&f| f > 0) + else { + return; + }; + let recovered_end = frontier - 1; + if self.should_increment_offset && self.offset.load(Ordering::Acquire) >= recovered_end { + return; + } + tracing::info!( - namespace_raw = self.consensus().namespace(), ++ namespace_raw = self.consensus().group(), + offset_frontier = frontier, + "restored partition offset frontier from its superblock" + ); + self.offset.store(recovered_end, Ordering::Release); + self.dirty_offset.store(recovered_end, Ordering::Relaxed); + self.should_increment_offset = true; + self.stats.set_current_offset(recovered_end); + } + + /// The next message offset this replica will mint, `0` while the offset + /// space is still empty. The value stamped into the durable record. + #[must_use] + pub fn offset_frontier(&self) -> u64 { + if self.should_increment_offset { + self.offset.load(Ordering::Acquire).saturating_add(1) + } else { + 0 + } + } + + /// Force the durable record to catch up with the current offset frontier, + /// outside the view-change gate. + /// + /// [`Self::persist_superblock_if_needed`] fires on `(view, log_view)` + /// changes only, which is the right trigger for the split-brain fence and + /// the wrong one for the frontier: an install can move the counter by + /// millions without touching the view. Called where the frontier changes + /// with nothing else durable naming it -- after a state-transfer install + /// and after the convergence that follows a failed one. Returns whether the + /// record now holds it; a failure is logged by the writer and left to the + /// ordinary retry, since the install itself already succeeded. + #[allow(clippy::future_not_send)] + #[must_use = "the bool is the durability verdict; dropping it silently ignores a failed write"] + pub async fn persist_offset_frontier(&self) -> bool { + self.persist_offset_frontier_at(self.offset_frontier()) + .await + } + + /// Record a frontier that may be LOWER than the one already on disk. + /// + /// The frontier is conditionally monotone: it advances everywhere except a + /// purge, which legitimately resets the offset space to 0. The advancing + /// form cannot express that -- it maxes against the live counter -- and the + /// distinction has to be explicit: a purge that leaves the old frontier + /// recorded makes the next boot re-seed the counter to the state the purge + /// just erased, and the following append stamps `base_offset` N where every + /// peer stamps 0. + #[allow(clippy::future_not_send)] + #[must_use = "the bool is the durability verdict; dropping it silently ignores a failed write"] + pub async fn reset_offset_frontier(&self) -> bool { + self.reset_offset_frontier_at(self.offset_frontier()).await + } + + /// [`Self::reset_offset_frontier`] for a frontier the live counter does not + /// hold yet. + /// + /// Two callers need the value spelled out rather than read off the counter. + /// A purge records its reset BEFORE it unlinks anything, while the counter + /// still names the pre-purge space, so a crash mid-unlink cannot boot into + /// a re-seed of the space the purge was erasing. An install under an + /// advancing purge generation records the offer's frontier, which is + /// legitimately below the local counter: the advancing form would max it + /// straight back up and leave the pre-purge value on disk across the swap + /// window. + #[allow(clippy::future_not_send)] + #[must_use = "the bool is the durability verdict; dropping it silently ignores a failed write"] + pub async fn reset_offset_frontier_at(&self, frontier: u64) -> bool { + let Some(superblock) = self.superblock.as_ref().map(Rc::clone) else { + return true; + }; + if self.superblock_write_is_backed_off() { + return false; + } + let _superblock_guard = self.superblock_lock.acquire().await; + self.write_superblock_inner(superblock.as_ref(), frontier) + .await + } + + /// Record the frontier immediately ahead of an irreversible quarantine, + /// BYPASSING the retry backoff. + /// + /// The gate exists because the other writers' callers became retry loops, + /// and skipping a doomed write costs them nothing. This caller is the + /// opposite: it writes once and then moves the segments that are the + /// record's only corroborating witness into `.fenced.N`, so a skip here is + /// not deferred work, it is the last chance gone. A disk that recovered + /// inside the backoff window would otherwise leave the rebuild re-seeding + /// from a stale record with nothing left to take the max against. + /// + /// `intended` is the frontier the caller knows the group is at, written + /// verbatim; `None` means the live counter is authoritative and the + /// advancing form applies. + #[allow(clippy::future_not_send)] + #[must_use = "the bool is the durability verdict; dropping it silently ignores a failed write"] + pub async fn record_frontier_before_quarantine(&self, intended: Option<u64>) -> bool { + let Some(superblock) = self.superblock.as_ref().map(Rc::clone) else { + return true; + }; + let _superblock_guard = self.superblock_lock.acquire().await; + match intended { + Some(frontier) => { + self.write_superblock_inner(superblock.as_ref(), frontier) + .await + } + None => { + self.write_superblock(superblock.as_ref(), self.offset_frontier()) + .await + } + } + } + + /// Whether a recent write failure's backoff window is still open. + /// + /// The same gate [`Self::persist_superblock_if_needed`] applies before its + /// own write, extended to the spelled-value writers because their callers + /// became retry loops: a deferred purge is re-issued by the reconciler, and + /// without this each pass re-runs a full `atomic_replace` against a disk + /// that just refused one, as fast as `ENOSPC` returns. + fn superblock_write_is_backed_off(&self) -> bool { + self.consensus.clock_realtime_micros() < self.superblock_retry_after_micros.get() + } + + /// [`Self::persist_offset_frontier`] for a frontier this replica has not + /// reached yet. + /// + /// Used to record an INCOMING frontier before a destructive swap: the + /// install unlinks the old chain and fsyncs that before the first staged + /// rename lands, and boot sweeps `.log.staging` unconditionally, so a crash + /// in that window otherwise leaves no copy of the frontier anywhere. Writing + /// the claim first makes it a durable lower bound the whole way through, and + /// over-claiming is harmless: the convergence that follows a failed install + /// seeds the counter from the same artifact frontier. + #[allow(clippy::future_not_send)] + #[must_use = "the bool is the durability verdict; dropping it silently ignores a failed write"] + pub async fn persist_offset_frontier_at(&self, frontier: u64) -> bool { + let Some(superblock) = self.superblock.as_ref().map(Rc::clone) else { + return true; + }; + if self.superblock_write_is_backed_off() { + return false; + } + let _superblock_guard = self.superblock_lock.acquire().await; + self.write_superblock(superblock.as_ref(), frontier).await + } + /// Burn one transfer stall round; `true` once the budget is exhausted. /// Lives on the partition, not the session, so a re-minted session /// cannot reset it (see [`Self::transfer_attempts`]). @@@ -2511,10 -2854,13 +2854,13 @@@ wher // reply. Emitting it would push an unrequested frame onto a real // client's lockstep reply stream if the sentinel ever routed there. if send_client_replies && !is_auto_commit_client(prepare_header.client) { - let reply = build_reply_message( - &prepare_header, - &committed_reply_body(prepare_header.operation), - ); + let body = match prepare_header.operation { + Operation::SendMessages => { - send_messages_reply_body(prepare_header.namespace, batch_stats) ++ send_messages_reply_body(prepare_header.group, batch_stats) + } + operation => committed_reply_body(operation), + }; + let reply = build_reply_message(&prepare_header, &body); let reply_buffers = reply.into_generic().into_frozen(); emit_sim_event(SimEventKind::ClientReplyEmitted, &event); diff --cc core/partitions/src/state_transfer.rs index d529a973f,833f07aba..a7b2deb20 --- a/core/partitions/src/state_transfer.rs +++ b/core/partitions/src/state_transfer.rs @@@ -1403,9 -2067,36 +2067,36 @@@ wher // that resumed minting at 0 would fork its batch stamps from // the group. A convergence failure outranks the install error: // the partition cannot serve and the caller must fence it. - self.converge_to_empty_after_failed_install(config, offsets_wire.next_offset) - .await - .map_err(|source| PartitionInstallError::ConvergeFailed { source })?; + self.converge_to_empty_after_failed_install( + config, + offsets_wire.next_offset, + staged_was_empty, + ) + .await + .map_err(|source| PartitionInstallError::ConvergeFailed { + source, + frontier: offsets_wire.next_offset, + })?; + } + // The frontier just moved with nothing durable naming it (an all-GC'd + // origin leaves no segment carrying it, and the crash windows inside + // the swap leave none either), so record it before returning. Runs for + // the converge path too: it seeds the counter from the same artifact. + // + // Logged rather than refused: the install already mutated, and the + // pre-swap write above left a valid lower bound on disk either way. The + // failure still matters -- the ordinary retry is the view-change gate, + // which an idle group may not reach for a long time -- so it must not + // pass silently. + if !self.persist_offset_frontier().await { + tracing::error!( + target: "iggy.partitions.diag", + plane = "partitions", - namespace_raw = self.consensus().namespace(), ++ namespace_raw = self.consensus().group(), + frontier = self.offset_frontier(), + "state-transfer install could not record the installed offset frontier; \ + the durable record stays at the pre-swap claim until the next view change" + ); } outcome } @@@ -1667,7 -2401,22 +2401,22 @@@ paths }; for path in old_consumer_paths.into_iter().chain(old_group_paths) { - let _ = delete_persisted_offset(&path).await; + if let Err(error) = delete_persisted_offset(&path).await { + // Not fatal, but not silent either: a stranded file is an id + // absent from the NEW table (matching ids get overwritten at + // the same path), and boot resurrects it. Sharpest after a + // purged origin ships `next_offset = 0`, where the clamp drops + // every incoming entry and the whole old table survives while + // the install still reports success. + tracing::warn!( + target: "iggy.partitions.diag", + plane = "partitions", - namespace_raw = self.consensus().namespace(), ++ namespace_raw = self.consensus().group(), + path = %path, + %error, + "failed to unlink a superseded consumer-offset file during install" + ); + } } self.persisted_offsets.borrow_mut().clear(); self.pending_consumer_offset_commits.clear(); diff --cc core/server-ng/src/dispatch.rs index 6e18ace6f,cbae3f257..2004813bd --- a/core/server-ng/src/dispatch.rs +++ b/core/server-ng/src/dispatch.rs @@@ -1398,9 -1411,10 +1422,10 @@@ async fn handle_default_non_replicated< shard: &Rc<ShellShard<B, MJ, S, SB>>, transport_client_id: u128, code: u32, - request: &Message<RequestHeader>, + request: &Message<RoutedRequestHeader>, user_id: Option<u32>, roster: &ClusterRoster, + client_ip: Option<IpAddr>, ) where B: ShellBus, MJ: JournalHandle + 'static, diff --cc core/server-ng/src/http/submit.rs index 0843daa2e,07c45b182..67f34aac3 --- a/core/server-ng/src/http/submit.rs +++ b/core/server-ng/src/http/submit.rs @@@ -26,8 -26,9 +26,9 @@@ use bytes::Bytes use consensus::MetadataHandle; use futures::channel::oneshot; use iggy_binary_protocol::consensus::Command2; - use iggy_binary_protocol::{GenericHeader, Operation, RoutedRequestHeader}; -use iggy_binary_protocol::{GenericHeader, Operation, ReplyHeader, RequestHeader}; ++use iggy_binary_protocol::{GenericHeader, Operation, ReplyHeader, RoutedRequestHeader}; use iggy_common::IggyError; + use message_bus::BusMessage; use metadata::impls::metadata::StreamsFrontend; use server_common::Message; use tracing::warn; diff --cc core/server-ng/src/partition_helpers.rs index 273a0c055,020c5155d..97ef435db --- a/core/server-ng/src/partition_helpers.rs +++ b/core/server-ng/src/partition_helpers.rs @@@ -519,6 -537,16 +537,16 @@@ pub(crate) fn restore_partition_view consensus: &mut VsrConsensus<Rc<IggyMessageBus>>, state: &VsrState, ) { + // The one line proving the durable record was READ BACK, not merely written: + // the group's whole anti-regression guarantee rests on this call running, and + // a replica that came back at view 0 is otherwise indistinguishable from one + // that resumed correctly until it votes. + info!( - namespace_raw = consensus.namespace(), ++ namespace_raw = consensus.group(), + view = state.view, + log_view = state.log_view, + "restored partition view from its superblock" + ); consensus.set_view(state.view); consensus.set_log_view(state.log_view); consensus.mark_superblock_durable(state.view, state.log_view); diff --cc core/server_common/src/consensus_message.rs index 7b3cf8a57,4fe65638b..5e0d34c71 --- a/core/server_common/src/consensus_message.rs +++ b/core/server_common/src/consensus_message.rs @@@ -656,7 -654,7 +656,7 @@@ wher #[cfg(test)] mod tests { use super::*; -- use iggy_binary_protocol::{Operation, ReplyHeader}; ++ use iggy_binary_protocol::{Operation, ReplyHeader, RequestHeader}; use smallvec::smallvec; // Field offsets via `offset_of!`: a field reorder fails to compile here @@@ -820,7 -815,7 +820,10 @@@ #[test] fn try_as_typed_invalid_validation_returns_err() { - // RoutedRequestHeader::validate rejects operation=Register with non-zero session. - // RequestHeader::validate rejects operation=Register with non-zero session. ++ // `RequestHeader::validate` rejects operation=Register with non-zero ++ // session. The client-wire header is the one carrying the field rules; ++ // `RoutedRequestHeader` is the post-resolution shape and gates on the ++ // command alone. let mut owned = header_bytes(Command2::Request, 256); { let buf = owned.as_mut_slice(); @@@ -902,8 -897,8 +905,11 @@@ } #[test] -- fn messagebag_dispatch_request_with_invalid_register_session_returns_err() { - // `RoutedRequestHeader::validate` rejects Register with non-zero session. ++ fn client_wire_decode_of_request_with_invalid_register_session_returns_err() { + // `RequestHeader::validate` rejects Register with non-zero session. ++ // Not routed through `MessageBag`: that path decodes the already-routed ++ // `RoutedRequestHeader` for replica-to-replica frames, and the field ++ // rules are the client boundary's job. let mut owned = header_bytes(Command2::Request, 256); { let buf = owned.as_mut_slice(); @@@ -911,16 -906,16 +917,16 @@@ buf[REQUEST_SESSION_OFF..REQUEST_SESSION_OFF + 8].copy_from_slice(&5u64.to_le_bytes()); } let generic = Message::<GenericHeader>::try_from(owned).expect("valid generic"); -- let result = MessageBag::try_from(generic); ++ let result = generic.try_into_typed::<RequestHeader>(); assert!(matches!(result, Err(ConsensusError::InvalidField(_)))); } // Ingress validation runs on every client frame at the network boundary, -- // reached through `MessageBag::try_from` -> `try_into_typed` -> - // `RoutedRequestHeader::validate`. Several dedup and authz conclusions rest on - // `RequestHeader::validate`. Several dedup and authz conclusions rest on -- // it running, so pin the field rules rather than the plumbing: whatever -- // `request_preflight` and the operation gate see downstream has already -- // passed these. ++ // reached through `try_into_typed` -> `RequestHeader::validate` before ++ // dispatch promotes the frame to `RoutedRequestHeader`. Several dedup and ++ // authz conclusions rest on it running, so pin the field rules rather than ++ // the plumbing: whatever `request_preflight` and the operation gate see ++ // downstream has already passed these. #[test] fn ingress_validation_enforces_the_request_header_field_rules() { // (operation, session, request, must_pass) @@@ -951,7 -946,7 +957,7 @@@ .copy_from_slice(&request.to_le_bytes()); } let generic = Message::<GenericHeader>::try_from(owned).expect("valid generic"); -- let accepted = MessageBag::try_from(generic).is_ok(); ++ let accepted = generic.try_into_typed::<RequestHeader>().is_ok(); assert_eq!( accepted, must_pass, "{operation:?} with session={session} request={request}" @@@ -973,7 -968,7 +979,7 @@@ } let generic = Message::<GenericHeader>::try_from(owned).expect("valid generic"); assert!(matches!( -- MessageBag::try_from(generic), ++ generic.try_into_typed::<RequestHeader>(), Err(ConsensusError::InvalidField(_)) )); } diff --cc core/shard/src/lib.rs index 1c3a49950,5563f219c..461308007 --- a/core/shard/src/lib.rs +++ b/core/shard/src/lib.rs @@@ -2930,7 -3201,7 +3201,7 @@@ wher } /// Resolve the single partition a VSR control frame addresses, keyed by -- /// `header.namespace`. Warns and returns `None` when the namespace matches ++ /// `header.group`. Warns and returns `None` when the namespace matches /// neither metadata nor a live partition consensus. Returns `&mut` because /// `on_do_view_change` / `on_commit` need it for `commit_journal`; the read- /// only callers reborrow `&`. Pump-only (sole mutator), so the `&mut` formed @@@ -3182,10 -3453,22 +3453,21 @@@ let config = planes.1.0.config(); // Counted BEFORE the `&mut partition` below exists: the scan takes // shared borrows of every partition (see `arm_partition_transfer`). - let transfers_inflight = self.partition_transfers_inflight(); + // Gated on the arm actually being possible, so a stale or misdirected + // frame -- and every StartView for a group that is not awaiting a + // transfer, which is all of them during an ordinary view change -- does + // not pay a node-wide scan. (A shard-level counter would remove the scan + // entirely, but `IggyPartition::transfer` is `pub` and cleared inside the + // partitions crate, so an externally maintained count would drift; that + // refactor is a prerequisite, not a detail.) - let transfers_inflight = if Self::may_arm_partition_transfer(&planes.1.0, header.namespace) - { ++ let transfers_inflight = if Self::may_arm_partition_transfer(&planes.1.0, header.group) { + self.partition_transfers_inflight() + } else { + 0 + }; let Some(partition) = self.resolve_partition_target( &planes.1.0, - header.namespace, + header.group, header.view, header.replica, "StartView", @@@ -3627,16 -3905,24 +3904,24 @@@ }); let header = *msg.header(); let planes = self.plane.inner(); - // `header.namespace == 0`: pre-upgrade metadata WAL entries were - // journaled before prepares stamped `consensus.group()`, and - // repair ships stored bytes verbatim -- without this acceptance a - // mixed-version metadata repair re-ships the same 0-stamped entries - // forever. Safe to claim for the metadata plane: partition ops never - // reach this arm with a metadata operation code, and namespace 0's - // partition-side gates journal nothing for foreign operations. + // Legacy acceptance: pre-upgrade metadata WAL entries were journaled - // before prepares stamped `consensus.namespace()`, and repair ships ++ // before prepares stamped `consensus.group()`, and repair ships + // stored bytes verbatim, so without it a mixed-version metadata repair + // re-ships the same 0-stamped entries forever. + // + // Keyed on the OPERATION, not on whether partition 0/0/0 exists: raw + // namespace 0 is `IggyNamespace::new(0, 0, 0)` and ids slab-allocate + // from 0, so 0/0/0 is the first partition every cluster creates -- on a + // single-shard node a "no partition 0 materialised" conjunct goes false + // the moment one topic exists and disables this migration exactly where + // it is needed. `is_metadata_plane` is the plane's OWN applicability + // predicate (the session ops `Register`/`Logout` replicate here without + // being metadata mutations, so `is_metadata` alone is too narrow), which + // is why both sites share it rather than re-deriving the set. + let metadata_plane_op = header.operation.is_metadata_plane(); - let legacy_metadata_claim = header.namespace == 0 && metadata_plane_op; ++ let legacy_metadata_claim = header.group == 0 && metadata_plane_op; if let Some(ref consensus) = planes.0.consensus - && (consensus.group() == header.group - || (header.group == 0 && !header.operation.is_partition())) - && (consensus.namespace() == header.namespace || legacy_metadata_claim) ++ && (consensus.group() == header.group || legacy_metadata_claim) { let session = *self.metadata_repair.borrow(); let Some(session) = session else { @@@ -3677,6 -3963,25 +3962,25 @@@ consensus.set_last_prepare_checksum(header.checksum); return; } + // A metadata-plane op that did not match above (no metadata consensus on + // this shard, or a namespace neither plane claims) is DROPPED, never + // offered to the partition arm. Falling through let a metadata prepare + // reach `apply_repaired_prepare`: it journals nothing, but it resets the + // partition repair session's idle ticks (masking a genuine stall) and + // carries the metadata prepare's checksum into the partition consensus + // via `set_last_prepare_checksum` -- inert only while prepare checksums + // are structurally zero, and a cross-plane parent stamp the moment the + // checksum chain is activated (see the note in `consensus::impls`). + if metadata_plane_op { + tracing::debug!( + shard = self.id, + op = header.op, + operation = ?header.operation, - namespace_raw = header.namespace, ++ namespace_raw = header.group, + "dropping a metadata-plane repair prepare this shard cannot journal" + ); + return; + } let Some(partition) = planes .1 .0 @@@ -3795,9 -4100,15 +4099,14 @@@ } return; } - // Counted BEFORE the `&mut partition` below exists: the scan takes - // shared borrows of every partition (see `arm_partition_transfer`). - let transfers_inflight = self.partition_transfers_inflight(); + // Counted BEFORE the `&mut partition` below exists, and only when an arm + // is possible at all: see the StartView site for why the scan is gated + // rather than replaced with a counter. - let transfers_inflight = if Self::may_arm_partition_transfer(&planes.1.0, header.namespace) - { ++ let transfers_inflight = if Self::may_arm_partition_transfer(&planes.1.0, header.group) { + self.partition_transfers_inflight() + } else { + 0 + }; let config = planes.1.0.config().clone(); let Some(partition) = planes .1 @@@ -4109,9 -4431,15 +4429,15 @@@ h.cluster = cluster; h.replica = self_id; h.nonce = nonce; - h.namespace = namespace; + h.group = namespace; h.size = total_size as u32; - if let Some((_, commit_op)) = offer { + // The serving replica's own progress travels with every descriptor, + // available or not: it is what lets a receiver refuse an offer from + // a replica that knows less than it does. + h.view = descriptor.view; + h.commit_max = descriptor.commit_max; + h.unavailable_transient = u8::from(descriptor.transient); + if let Some((_, commit_op)) = descriptor.offer { h.available = 1; h.commit_op = commit_op; } @@@ -4222,8 -4550,13 +4548,13 @@@ self_id, header.replica, header.nonce, - header.namespace, + header.group, - Some((&offer.manifest(), offer.commit_op)), + TransferDescriptor::available( + &offer.manifest(), + offer.commit_op, + consensus.view(), + consensus.commit_max(), + ), ) .await; return; @@@ -4245,8 -4578,13 +4576,13 @@@ self_id, header.replica, header.nonce, - header.namespace, + header.group, - Some((&offer.manifest(), offer.commit_op)), + TransferDescriptor::available( + &offer.manifest(), + offer.commit_op, + consensus.view(), + consensus.commit_max(), + ), ) .await; self.state_transfer_offers.borrow_mut().insert( @@@ -4275,8 -4613,12 +4611,12 @@@ self_id, header.replica, header.nonce, - header.namespace, + header.group, - None, + TransferDescriptor::unavailable( + false, + consensus.view(), + consensus.commit_max(), + ), ) .await; } @@@ -4626,61 -4971,64 +4969,64 @@@ let reply = { let mut offers = self.state_transfer_offers.borrow_mut(); let served = offers - .get_mut(&(header.namespace, header.replica)) + .get_mut(&(header.group, header.replica)) .filter(|served| served.nonce == header.nonce); - served.map_or(Some(ChunkReply::UnknownOffer), |served| { - let ServedOffer::Metadata(offer) = &served.offer else { - return Some(ChunkReply::UnknownOffer); - }; - // Manifest-index addressing: an index past the offer is a - // requester bug (or a stale frame) and is dropped below. - let last_artifact = offer.len().saturating_sub(1); - let artifact_bytes = offer.payload(header.artifact as usize)?; - let start = header.offset as usize; - // A request AT the end of an artifact has nothing left to serve. - // Answering it with `Some(&[])` -- which `get(len..len)` happily - // returns -- would extend nothing on the receiver, reset both - // sides' idle counters, and be re-requested at the same offset - // forever: an unbounded empty-frame ping-pong with the rejoining - // replica withholding `PrepareOk` for the life of the process. - // Reachable when a rebuilt offer is SHORTER than the manifest the - // receiver accepted (a client logged out between the two builds). - if start >= artifact_bytes.len() { - return None; - } - let end = start - .saturating_add((header.len as usize).min(chunk_len_max)) - .min(artifact_bytes.len()); - let payload = artifact_bytes.get(start..end)?; - // Only now that bytes are actually going out: an out-of-bounds or - // stale frame must not flip a live offer onto the short expiry. - // Tail of the final artifact means the receiver holds everything - // the manifest promised, so the offer only has to outlive a - // possible re-request of this very chunk. - if header.artifact as usize == last_artifact && end >= artifact_bytes.len() { - served.fully_served = true; - } - // Serving a chunk is the only liveness signal the offer gets; - // the expiry sweep drops it once these stop arriving. Set here - // rather than on entry so a request that serves NOTHING cannot - // keep an abandoned offer alive. - served.idle_ticks = 0; - let total_size = size_of::<StateChunkHeader>() + payload.len(); - let mut chunk = Message::<StateChunkHeader>::new(total_size); - chunk.as_mut_slice()[size_of::<StateChunkHeader>()..].copy_from_slice(payload); - Some(ChunkReply::Chunk(chunk.transmute_header( - |_, h: &mut StateChunkHeader| { - h.command = Command2::StateChunk; - h.cluster = cluster; - h.replica = self_id; - h.nonce = header.nonce; - h.group = header.group; - h.artifact = header.artifact; - h.offset = header.offset; - h.size = total_size as u32; - }, - ))) - }) + served.map_or( + Some(ChunkReply::Unavailable { transient: true }), + |served| { + let ServedOffer::Metadata(offer) = &served.offer else { + return Some(ChunkReply::Unavailable { transient: true }); + }; + // Manifest-index addressing: an index past the offer is a + // requester bug (or a stale frame) and is dropped below. + let last_artifact = offer.len().saturating_sub(1); + let artifact_bytes = offer.payload(header.artifact as usize)?; + let start = header.offset as usize; + // A request AT the end of an artifact has nothing left to serve. + // Answering it with `Some(&[])` -- which `get(len..len)` happily + // returns -- would extend nothing on the receiver, reset both + // sides' idle counters, and be re-requested at the same offset + // forever: an unbounded empty-frame ping-pong with the rejoining + // replica withholding `PrepareOk` for the life of the process. + // Reachable when a rebuilt offer is SHORTER than the manifest the + // receiver accepted (a client logged out between the two builds). + if start >= artifact_bytes.len() { + return None; + } + let end = start + .saturating_add((header.len as usize).min(chunk_len_max)) + .min(artifact_bytes.len()); + let payload = artifact_bytes.get(start..end)?; + // Only now that bytes are actually going out: an out-of-bounds or + // stale frame must not flip a live offer onto the short expiry. + // Tail of the final artifact means the receiver holds everything + // the manifest promised, so the offer only has to outlive a + // possible re-request of this very chunk. + if header.artifact as usize == last_artifact && end >= artifact_bytes.len() { + served.fully_served = true; + } + // Serving a chunk is the only liveness signal the offer gets; + // the expiry sweep drops it once these stop arriving. Set here + // rather than on entry so a request that serves NOTHING cannot + // keep an abandoned offer alive. + served.idle_ticks = 0; + let total_size = size_of::<StateChunkHeader>() + payload.len(); + let mut chunk = Message::<StateChunkHeader>::new(total_size); + chunk.as_mut_slice()[size_of::<StateChunkHeader>()..].copy_from_slice(payload); + Some(ChunkReply::Chunk(chunk.transmute_header( + |_, h: &mut StateChunkHeader| { + h.command = Command2::StateChunk; + h.cluster = cluster; + h.replica = self_id; + h.nonce = header.nonce; - h.namespace = header.namespace; ++ h.group = header.group; + h.artifact = header.artifact; + h.offset = header.offset; + h.size = total_size as u32; + }, + ))) + }, + ) }; match reply { Some(ChunkReply::Chunk(chunk)) => { @@@ -4700,8 -5049,12 +5047,12 @@@ self_id, header.replica, header.nonce, - header.namespace, + header.group, - None, + TransferDescriptor::unavailable( + transient, + consensus.view(), + consensus.commit_max(), + ), ) .await; } @@@ -5286,86 -5728,155 +5726,155 @@@ let cached = self .state_transfer_offers .borrow_mut() - .get_mut(&(header.namespace, header.replica)) + .get_mut(&(header.group, header.replica)) .filter(|served| served.nonce == header.nonce) .and_then(|served| { - let ServedOffer::Partition { offer, .. } = &served.offer else { + let ServedOffer::Partition(offer) = &served.offer else { return None; }; served.idle_ticks = 0; Some(Rc::clone(offer)) }); - if let Some(offer) = cached { - tracing::debug!( - shard = self.id, - namespace_raw = header.group, - requester = header.replica, - "re-answering a partition state transfer request from the offer already served" - ); - self.send_state_transfer_target( - cluster, - self_id, - header.replica, - header.nonce, - header.group, - Some((&offer.manifest(), offer.commit_op)), - ) - .await; - return; - } - match partition.state_transfer_offer(&config).await { - Ok(offer) => { - tracing::info!( + // One resolve, one send: the three outcomes differ only in the + // descriptor they produce, and duplicating the send made it possible for + // them to drift on the progress they advertise. + let offer = match cached { + Some(offer) => { + tracing::debug!( shard = self.id, - namespace_raw = header.namespace, + namespace_raw = header.group, requester = header.replica, - commit_op = offer.commit_op, - artifacts = offer.artifact_count(), - total_len = offer.total_len(), - "serving partition state transfer" - ); - self.send_state_transfer_target( - cluster, - self_id, - header.replica, - header.nonce, - header.group, - Some((&offer.manifest(), offer.commit_op)), - ) - .await; - self.state_transfer_offers.borrow_mut().insert( - (header.group, header.replica), - ServedStateTransfer { - nonce: header.nonce, - offer: ServedOffer::Partition { offer }, - idle_ticks: 0, - fully_served: false, - }, + "re-answering a partition state transfer request from the offer \ + already served" ); + Some(offer) } - Err(reason) => { - // The ACTUAL reason: "not the caught-up primary" is routine - // (the requester re-targets), an unreadable segment is an - // operator-visible fault on THIS node. - None if !self.may_serve_another_partition_transfer(header.namespace) => { ++ None if !self.may_serve_another_partition_transfer(header.group) => { + // Admission control, because the served-payload budget is a + // BYTE budget and the pulls that overrun it do not degrade + // gracefully. Each concurrent pull holds a different segment + // resident, so admitting more distinct groups than the budget + // has max-size slots makes them evict each other on every + // chunk: every request then re-reads and re-hashes a whole + // segment to serve one 256 KiB range, and a per-chunk serve + // that outruns the requester's stall interval exhausts its + // retry budget, so the pull rotates peers and never converges. + // Refusing the surplus is what makes the admitted ones finish. tracing::info!( shard = self.id, - namespace_raw = header.namespace, + namespace_raw = header.group, requester = header.replica, - %reason, - "cannot serve partition state transfer; requester falls back" + "already serving as many partition transfers as the served-payload \ + budget holds; refusing until one completes" ); + let (view, commit_max) = serving_progress(partition); self.send_state_transfer_target( cluster, self_id, header.replica, header.nonce, - header.namespace, + header.group, - None, + TransferDescriptor::unavailable(true, view, commit_max), ) .await; + return; } - } + None => { + // Claim the admission slot for the whole build, not just for a + // completed offer: the checksum pass runs over several rounds + // and holds nothing in the offers map meanwhile. + self.partition_offer_builds + .borrow_mut() - .insert(header.namespace, 0); ++ .insert(header.group, 0); + match partition.state_transfer_offer(&config).await { + Ok(offer) => { + self.partition_offer_builds + .borrow_mut() - .remove(&header.namespace); ++ .remove(&header.group); + tracing::info!( + shard = self.id, - namespace_raw = header.namespace, ++ namespace_raw = header.group, + requester = header.replica, + commit_op = offer.commit_op, + artifacts = offer.artifact_count(), + total_len = offer.total_len(), + "serving partition state transfer" + ); + self.state_transfer_offers.borrow_mut().insert( - (header.namespace, header.replica), ++ (header.group, header.replica), + ServedStateTransfer { + nonce: header.nonce, + offer: ServedOffer::Partition(Rc::clone(&offer)), + idle_ticks: 0, + fully_served: false, + }, + ); + Some(offer) + } + Err(reason) => { + // The ACTUAL reason: "not the caught-up primary" is routine + // (the requester re-targets), an unreadable segment is an + // operator-visible fault on THIS node. The requester cannot + // see the reason, only whether it was transient, which is + // what keeps a routine refusal from charging its failure + // count. + // + // The slot survives ONLY a budget-exhausted round, which is + // a build that will resume; every other refusal abandons + // the build and must not keep the group admitted. + let building = matches!( + reason, + partitions::state_transfer::PartitionTransferUnavailable::OfferBuildInProgress { .. } + ); + if !building { + self.partition_offer_builds + .borrow_mut() - .remove(&header.namespace); ++ .remove(&header.group); + } + let transient = reason.transient(); + tracing::info!( + shard = self.id, - namespace_raw = header.namespace, ++ namespace_raw = header.group, + requester = header.replica, + transient, + %reason, + "cannot serve partition state transfer; requester falls back" + ); + let (view, commit_max) = serving_progress(partition); + self.send_state_transfer_target( + cluster, + self_id, + header.replica, + header.nonce, - header.namespace, ++ header.group, + TransferDescriptor::unavailable(transient, view, commit_max), + ) + .await; + return; + } + } + } + }; + let Some(offer) = offer else { + return; + }; + // Sampled AFTER any build: that build force-flushes and hashes a + // budgeted slice of the un-memoized segments (a first multi-GiB serve + // spans several rounds before an offer exists) while reading + // `commit_op` post-flush, so a pre-build sample could advertise a + // `commit_max` below the descriptor's own `commit_op` -- which only + // makes the receiver's gate refuse, and refusals feed a backoff. + let (view, commit_max) = serving_progress(partition); + self.send_state_transfer_target( + cluster, + self_id, + header.replica, + header.nonce, - header.namespace, ++ header.group, + TransferDescriptor::available(&offer.manifest(), offer.commit_op, view, commit_max), + ) + .await; } /// Serve one partition chunk. Segment payloads are loaded from disk on @@@ -5396,8 -5907,16 +5905,12 @@@ } let header = *msg.header(); + // The requester id keys the offer map, whose bound is the replica count. + if !self.peer_is_known(header.replica, "RequestStateChunk") { + return; + } let planes = self.plane.inner(); - let Some(partition) = planes - .1 - .0 - .get_by_ns(&IggyNamespace::from_raw(header.namespace)) - else { + let Some(partition) = planes.1.0.get_by_ns(&IggyNamespace::from_raw(header.group)) else { return; }; let cluster = partition.consensus().cluster(); @@@ -5407,13 -5928,17 +5922,17 @@@ let attempt = 'attempt: { let mut offers = self.state_transfer_offers.borrow_mut(); let served = offers - .get_mut(&(header.namespace, header.replica)) + .get_mut(&(header.group, header.replica)) .filter(|served| served.nonce == header.nonce); let Some(served) = served else { - break 'attempt ChunkAttempt::Reply(Some(ChunkReply::UnknownOffer)); + break 'attempt ChunkAttempt::Reply(Some(ChunkReply::Unavailable { + transient: true, + })); }; - let ServedOffer::Partition { offer } = &served.offer else { - break 'attempt ChunkAttempt::Reply(Some(ChunkReply::UnknownOffer)); + let ServedOffer::Partition(offer) = &served.offer else { + break 'attempt ChunkAttempt::Reply(Some(ChunkReply::Unavailable { + transient: true, + })); }; let last_artifact = offer.artifact_count().saturating_sub(1); let artifact = header.artifact as usize; @@@ -5456,8 -5981,18 +5975,18 @@@ let Some(payload) = artifact_bytes.get(start..end) else { break 'attempt ChunkAttempt::Reply(None); }; - if artifact == last_artifact && end >= artifact_bytes.len() { + if artifact == last_artifact && end >= artifact_bytes.len() && !served.fully_served + { served.fully_served = true; + // Once per transfer, at the last byte of the last artifact. + // The descriptor log only proves a REQUEST arrived; this is + // the serving side's proof that the pull ran to completion. + tracing::info!( + shard = self.id, - namespace_raw = header.namespace, ++ namespace_raw = header.group, + requester = header.replica, + "partition state transfer fully served" + ); } served.idle_ticks = 0; let total_size = size_of::<StateChunkHeader>() + payload.len(); @@@ -5479,21 -6014,41 +6008,41 @@@ match attempt { ChunkAttempt::Reply(reply) => break reply, ChunkAttempt::Load { log_path, entry } => { - if let Some(bytes) = load_partition_artifact(&log_path, &entry).await { - self.served_segment_cache.borrow_mut().insert( - header.group, - entry.checksum, - Rc::new(bytes), - ); - continue; - } + // Chunked read + incremental hash: this runs on the pump to + // answer ONE 256 KiB chunk request, so a whole-file read + // plus a single hash pass over up to a segment would be one + // long uninterruptible CPU+IO block. The chunking keeps the + // REACTOR moving; this shard's consensus ticks are a sibling + // select arm of the same task and stay frozen either way. + let loaded = partitions::state_transfer::load_verified_segment_artifact( + &log_path, &entry, + ) + .await; + let reason = match loaded { + Ok(bytes) => { + self.served_segment_cache.borrow_mut().insert( - header.namespace, ++ header.group, + entry.checksum, + Rc::new(bytes), + self.served_segment_cache_bytes_max.get(), + ); + continue; + } + Err(reason) => reason, + }; + // The CAUSE decides what the requester is told: a racing GC + // or a stale offer is transient and costs it nothing, while + // an unreadable device is this node's fault and must charge, + // or a dying disk reads as a momentary blip forever. + let transient = reason.transient(); tracing::warn!( shard = self.id, - namespace_raw = header.namespace, + namespace_raw = header.group, artifact = header.artifact, path = %log_path, - "served segment no longer matches its manifest entry; \ - evicting the offer" + transient, + %reason, + "cannot serve the requested segment; evicting the offer" ); self.state_transfer_offers .borrow_mut() @@@ -5514,11 -6069,12 +6063,12 @@@ .send_to_replica(header.replica, chunk.into_generic().into_frozen()) .await; } - Some(ChunkReply::UnknownOffer) => { + Some(ChunkReply::Unavailable { transient }) => { tracing::info!( shard = self.id, - namespace_raw = header.namespace, + namespace_raw = header.group, requester = header.replica, + transient, "partition chunk request for an unknown offer; telling requester to restart" ); self.send_state_transfer_target( @@@ -5526,8 -6082,13 +6076,13 @@@ self_id, header.replica, header.nonce, - header.namespace, + header.group, - None, + // Usually TRANSIENT -- retention GC'd a served segment, or + // the offer aged out between two chunks, and the restarted + // session converges -- but a load that failed on a local + // fault says so, or a dying disk would read as a momentary + // blip forever. + TransferDescriptor::unavailable(transient, serving_view, serving_commit_max), ) .await; } @@@ -5705,22 -6446,91 +6440,91 @@@ return; } if header.available == 0 { - // Charged like any other failure: `is_caught_up_primary` on the - // serving side wants `commit_min == commit_max`, frequently - // false under produce load, so an uncharged immediate re-arm - // looped repair -> refusal -> arm -> unavailable at network - // round-trip rate. The backoff + peer rotation live in the - // shared re-arm path. + // A refusal the peer marked transient (it is momentarily not the + // caught-up primary, which `is_caught_up_primary` makes frequent + // under produce load) must not charge the consecutive-failure count: + // that count is reset only by a completed install, so ten routine + // refusals pin the re-arm backoff at its 1024x ceiling while nothing + // else recovers the partition -- repair keeps hitting the refused + // floor and will not arm while a re-arm is pending. A hard refusal + // (unreadable segment, failed flush) still charges. + let transient = header.unavailable_transient == 1; tracing::info!( shard = self.id, - namespace_raw = header.namespace, + namespace_raw = header.group, peer = header.replica, + transient, "partition transfer peer cannot serve; backing off before re-arming" ); + if transient { + // The peer that refused is the node that would otherwise serve, + // and on the partition arm only a caught-up primary can. Keep + // asking it unless it is not the primary this replica knows: a + // rotation spends the next round on a backup that can only + // refuse, and the serving side's partial offer-build progress + // is memoized per node, so that round advances no hashing. + let primary = { + let consensus = partition.consensus(); + consensus.primary_index(consensus.view()) + }; + self.rearm_partition_transfer_after_refusal( + partition, + header.replica, + header.replica != primary, + ) + .await; + } else { + self.abandon_or_rearm_partition_transfer(partition, header.replica) + .await; + } + return; + } + // The serving replica's own progress, carried by every descriptor: an + // offer from a replica that knows LESS than this one does is the phantom + // view-0 primary signature (a group whose directory vanished boots + // `init()`, comes up Normal at view 0, and an empty log is trivially + // caught up). Installing it would unlink a chain this replica already + // holds; nonce match alone cannot tell the two apart. + let local_view = partition.consensus().view(); + let local_commit_max = partition.consensus().commit_max(); + // `commit_op` past the sender's OWN `commit_max` is self-contradictory: + // the offer cannot be built past the frontier its builder had. Nothing + // downstream bounds it above -- the install only refuses values BELOW + // the local floor, and the offsets-artifact cross-check compares two + // numbers the same peer chose -- so without this a peer offering + // `commit_op = u64::MAX` drives this replica's commit floor, sequencer + // and `commit_max` there and it reports itself fully committed. + if header.commit_op > header.commit_max { + tracing::warn!( + shard = self.id, - namespace_raw = header.namespace, ++ namespace_raw = header.group, + peer = header.replica, + serving_commit_op = header.commit_op, + serving_commit_max = header.commit_max, + "refusing a partition transfer offer whose commit_op exceeds the sender's \ + own commit frontier" + ); self.abandon_or_rearm_partition_transfer(partition, header.replica) .await; return; } + if header.view < local_view || header.commit_max < local_commit_max { + tracing::warn!( + shard = self.id, - namespace_raw = header.namespace, ++ namespace_raw = header.group, + peer = header.replica, + serving_view = header.view, + serving_commit_max = header.commit_max, + local_view, + local_commit_max, + "refusing a partition transfer offer from a replica behind this one" + ); + // ALWAYS rotate: this refusal is evidence about the peer, not about + // its timing, so re-asking it is the one thing that cannot help. + self.rearm_partition_transfer_after_refusal(partition, header.replica, true) + .await; + return; + } let manifest_bytes = &msg.as_slice()[size_of::<StateTransferTargetHeader>()..header.size as usize]; let entries = match consensus::decode_state_manifest(manifest_bytes) { @@@ -5772,6 -6584,16 +6578,16 @@@ return; } let reused = partition.reuse_staged_segments(&entries).await; + if !reused.is_empty() { + tracing::info!( + shard = self.id, - namespace_raw = header.namespace, ++ namespace_raw = header.group, + peer = header.replica, + adopted = reused.len(), + artifacts = entries.len(), + "adopted staged segments from an earlier transfer attempt" + ); + } // Re-check the nonce AFTER the await: the staging scan yields, and a // session re-minted underneath it must not get stamped with this // (now stale) offer's commit_op and manifest. @@@ -6100,17 -6994,108 +6988,108 @@@ ) where B: MessageBus, { - partition.transfer = None; let failures = partition.record_transfer_failure(); + let after_ticks = transfer_rearm_backoff(self.repair_retry_ticks.get(), failures); + self.schedule_partition_transfer_rearm(partition, peer, failures, after_ticks, true) + .await; + } + + /// Re-arm after a refusal the serving peer marked TRANSIENT: schedule the + /// next attempt on a flat interval and charge nothing. + /// + /// "The peer is momentarily not the caught-up primary" is the common case + /// under produce load, and `transfer_failures` is reset only by a completed + /// install, so charging it turns a transient into a stall measured in re-arm + /// ceilings: nothing else recovers the partition meanwhile, since repair + /// keeps hitting the refused floor and will not arm while a re-arm is + /// pending. + /// + /// `rotate` belongs to the CALLER because the two refusal sites mean + /// opposite things by it. A peer saying "not right now" is the node that + /// would otherwise serve, so staying on it is right. This replica refusing + /// a descriptor from a peer that knows LESS than it does is the one case + /// where the peer is provably the wrong one, and rotating is the whole + /// remedy: a restarted primary comes back at `commit_max = 0` (the + /// partition journal is memory-only), so a rejoining backup would otherwise + /// pin itself to it at a flat interval until the group's next election. + #[allow(clippy::future_not_send)] + async fn rearm_partition_transfer_after_refusal( + &self, + partition: &mut IggyPartition<B, SB>, + peer: u8, + rotate: bool, + ) where + B: MessageBus, + { + // Deliberately NOT `transfer_rearm_backoff`: a flat interval, so a peer + // that spends a minute catching up costs a minute of retries rather than + // a climb to the 1024x ceiling. + let after_ticks = self.repair_retry_ticks.get(); + // The flat interval means a partition can sit here for hours without + // charging anything, so the ONLY operator signal is this count: it + // escalates the log level and feeds a metric, and it never touches the + // backoff. + let refusals = partition.record_transfer_refusal(); + self.metrics.record_partition_transfer_refusal(); + if refusals >= TRANSFER_REFUSALS_BEFORE_ESCALATION + && refusals.is_multiple_of(TRANSFER_REFUSALS_BEFORE_ESCALATION) + { + // Deliberately not phrased as "not rejoining": a serving primary + // building a large offer refuses one round per budget slice, so a + // healthy multi-GiB rejoin reaches this count while progressing + // normally. The descriptor carries no reason code, so this side + // cannot tell the two apart; the serving node's own logs can. + tracing::warn!( + shard = self.id, - namespace_raw = partition.consensus().namespace(), ++ namespace_raw = partition.consensus().group(), + peer, + refusals, + "partition state transfer has been refused {refusals} times in a row; the peer \ + may be building a large offer or rate-limiting concurrent transfers, or it may \ + be unable to serve at all -- check its logs before intervening" + ); + } + self.schedule_partition_transfer_rearm(partition, peer, 0, after_ticks, rotate) + .await; + } + + /// Drop the session, pick the next peer, and schedule the re-arm; shared by + /// the charged and uncharged paths. + /// + /// `rotate` is false where the refusing peer is the only one that could + /// have served: only a caught-up primary passes `is_caught_up_primary`, so + /// rotating off it asks a backup that can answer nothing but another + /// refusal, and the serving side's partial offer-build progress is memoized + /// PER NODE, so the round spent on the backup also advances no hashing. + #[allow(clippy::future_not_send)] + async fn schedule_partition_transfer_rearm( + &self, + partition: &mut IggyPartition<B, SB>, + peer: u8, + failures: u32, + after_ticks: u32, + rotate: bool, + ) where + B: MessageBus, + { + partition.transfer = None; let consensus = partition.consensus(); if consensus.state_transfer_stage() != consensus::StateTransferStage::Idle { consensus.set_state_transfer_stage(consensus::StateTransferStage::Idle); } - let next_peer = next_transfer_peer(consensus.replica(), peer, consensus.replica_count()); - let after_ticks = transfer_rearm_backoff(self.repair_retry_ticks.get(), failures); + let next_peer = if rotate { + next_transfer_peer( + consensus.replica(), + peer, + consensus.replica_count(), + consensus.primary_index(consensus.view()), + ) + } else { + peer + }; tracing::info!( shard = self.id, - namespace_raw = partition.consensus().namespace(), + namespace_raw = partition.consensus().group(), failures, next_peer, after_ticks, diff --cc core/simulator/src/workload/ops/mod.rs index 815976072,be965e910..df6ffed48 --- a/core/simulator/src/workload/ops/mod.rs +++ b/core/simulator/src/workload/ops/mod.rs @@@ -84,7 -84,7 +84,7 @@@ macro_rules! op_dispatch /// In-flight entry recorded on submit, removed on reply. /// -- /// `request_namespace` is the `header.namespace` the request was ++ /// `request_namespace` is the `header.group` the request was /// submitted with; the auditor cross-checks it against the /// reply's namespace so a misrouted reply cannot update the /// wrong VSR group's bookkeeping. diff --cc foreign/node/scripts/check-vsr-protocol.mjs index cdcd432e2,cdcd432e2..ebfcf66d2 --- a/foreign/node/scripts/check-vsr-protocol.mjs +++ b/foreign/node/scripts/check-vsr-protocol.mjs @@@ -31,11 -31,11 +31,9 @@@ const rustHeader, rustCommand, rustOperation, -- rustNamespace, rustProtocolCargo, nodeCodes, nodeHeader, -- nodeNamespace, nodeOperation, nodeRegister, ] = await Promise.all([ @@@ -44,11 -44,11 +42,9 @@@ read('core/binary_protocol/src/consensus/header.rs'), read('core/binary_protocol/src/consensus/command.rs'), read('core/binary_protocol/src/consensus/operation.rs'), -- read('core/binary_protocol/src/namespace.rs'), read('core/binary_protocol/Cargo.toml'), readNode('src/wire/command.code.ts'), readNode('src/wire/vsr/header.ts'), -- readNode('src/wire/vsr/namespace.ts'), readNode('src/wire/vsr/operation.ts'), readNode('src/wire/vsr/register.ts'), ]); @@@ -125,46 -125,46 +121,14 @@@ assert.deepEqual 'Node replicated code-to-operation map differs from Rust dispatch' ); --const rustNamespaceValue = (name) => Number( -- rustNamespace.match( -- new RegExp(`pub const ${name}: usize = ([0-9_]+);`) -- )?.[1].replaceAll('_', '') --); --const nodeNamespaceValue = (name) => Number( -- nodeNamespace.match( -- new RegExp(`const ${name} = ([0-9_]+);`) -- )?.[1].replaceAll('_', '') --); --const namespaceLimits = new Map( -- ['MAX_STREAMS', 'MAX_TOPICS', 'MAX_PARTITIONS'].map((name) => [ -- name, -- rustNamespaceValue(name) -- ]) --); --for (const [name, value] of namespaceLimits) { -- assert.ok(Number.isSafeInteger(value), `Rust ${name} was not found`); -- assert.equal( -- nodeNamespaceValue(name), -- value, -- `Node ${name} differs from Rust namespace layout` -- ); --} -- --const bitsRequired = (value) => BigInt(value).toString(2).length; --const expectedTopicShift = bitsRequired( -- namespaceLimits.get('MAX_PARTITIONS') - 1 --); --const expectedStreamShift = expectedTopicShift + -- bitsRequired(namespaceLimits.get('MAX_TOPICS') - 1); --const namespaceModule = await import( -- pathToFileURL(resolve(nodeRoot, 'dist/wire/vsr/namespace.js')).href --); --assert.equal( -- namespaceModule.packNamespace(1, 1, 1), -- (1n << BigInt(expectedStreamShift)) | -- (1n << BigInt(expectedTopicShift)) | -- 1n, -- 'Node namespace shifts differ from Rust namespace layout' ++// No namespace-packing parity to check: the client wire carries no routing ++// namespace, so the packing rules stay entirely server-side and this SDK has ++// nothing to mirror. What still matters is that the client never grows a ++// namespace field back -- the offset recomputation below catches that, since ++// reintroducing one would move every field after it. ++assert.ok( ++ !/namespace/i.test(nodeHeader.replace(/\/\*[\s\S]*?\*\/|\/\/.*/g, '')), ++ 'Node request header must not carry a namespace field' ); const rustEvictionBlock = diff --cc foreign/node/src/wire/vsr/header.test.ts index 91b92def0,91b92def0..a8095c744 --- a/foreign/node/src/wire/vsr/header.test.ts +++ b/foreign/node/src/wire/vsr/header.test.ts @@@ -32,7 -32,7 +32,6 @@@ describe('VSR request header', () => client, request: 0x0102030405060708n, operation: 2, -- namespace: 0x8877665544332211n, session: 0x1020304050607080n, nonReplicatedCode: 60_001 }); @@@ -53,10 -53,10 +52,6 @@@ 0x0102030405060708n ); assert.equal(header.readUInt8(REQUEST_OFFSET.operation), 2); -- assert.equal( -- header.readBigUInt64LE(REQUEST_OFFSET.namespace), -- 0x8877665544332211n -- ); assert.equal( header.readBigUInt64LE(REQUEST_OFFSET.session), 0x1020304050607080n @@@ -71,7 -71,7 +66,6 @@@ client: 1n, request: 0n, operation: 1, -- namespace: 1n << 63n, session: 0n }); const expected = Buffer.alloc(HEADER_SIZE); @@@ -79,7 -79,7 +73,6 @@@ expected.writeUInt8(Command2.Request, REQUEST_OFFSET.command); expected.writeBigUInt64LE(1n, REQUEST_OFFSET.client); expected.writeUInt8(1, REQUEST_OFFSET.operation); -- expected.writeBigUInt64LE(1n << 63n, REQUEST_OFFSET.namespace); assert.deepEqual(header, expected); }); @@@ -90,7 -90,7 +83,6 @@@ client: maximum << 64n | maximum, request: maximum, operation: 160, -- namespace: maximum, session: maximum }); assert.equal(header.readBigUInt64LE(REQUEST_OFFSET.request), maximum); diff --cc foreign/node/src/wire/vsr/header.ts index bdee4f270,bdee4f270..9c51217dd --- a/foreign/node/src/wire/vsr/header.ts +++ b/foreign/node/src/wire/vsr/header.ts @@@ -27,7 -27,7 +27,15 @@@ /** Size of every consensus header, both directions. */ export const HEADER_SIZE = 256; --/** `RequestHeader` field offsets the client writes. */ ++/** ++ * `RequestHeader` field offsets the client writes. ++ * ++ * The client wire carries no routing namespace: the server derives the ++ * consensus group (plane from `operation`, partition target from the payload) ++ * and stamps it into its own internal header. Everything that followed the ++ * removed field therefore sits eight bytes earlier than in the pre-derivation ++ * layout. ++ */ export const REQUEST_OFFSET = { size: 48, command: 60, @@@ -35,9 -35,9 +43,8 @@@ timestamp: 160, request: 168, operation: 176, -- namespace: 184, -- session: 192, -- reserved: 204 ++ session: 184, ++ reserved: 196 } as const; /** `ReplyHeader` field offsets the client reads. */ @@@ -45,8 -45,8 +52,7 @@@ export const REPLY_OFFSET = size: 48, command: 60, operation: 208, -- namespace: 216, -- status: 224 ++ status: 216 } as const; /** `EvictionHeader` field offsets the client reads. */ @@@ -98,8 -98,8 +104,6 @@@ export type RequestHeaderFields = request: bigint, /** `Operation` discriminant. */ operation: number, -- /** Routing namespace (u64). */ -- namespace: bigint, /** Bound session (u64), or 0n. */ session: bigint, /** Command code for `NonReplicated`, placed in `reserved[0..4]`. */ @@@ -109,7 -109,7 +113,7 @@@ const U64_MASK = 0xFFFFFFFFFFFFFFFFn; /** -- * Encodes a 256-byte request header. Only the seven fields the server reads ++ * Encodes a 256-byte request header. Only the six fields the server reads * are written; the checksums stay zero, matching the Rust SDK's contract * with the VSR server. */ @@@ -122,7 -122,7 +126,6 @@@ export const encodeRequestHeader = (fie header.writeBigUInt64LE(fields.client >> 64n, REQUEST_OFFSET.client + 8); header.writeBigUInt64LE(fields.request, REQUEST_OFFSET.request); header.writeUInt8(fields.operation, REQUEST_OFFSET.operation); -- header.writeBigUInt64LE(fields.namespace, REQUEST_OFFSET.namespace); header.writeBigUInt64LE(fields.session, REQUEST_OFFSET.session); if (fields.nonReplicatedCode !== undefined) header.writeUInt32LE(fields.nonReplicatedCode, REQUEST_OFFSET.reserved); diff --cc foreign/node/src/wire/vsr/index.ts index 5c079ba4f,5c079ba4f..ca88e7945 --- a/foreign/node/src/wire/vsr/index.ts +++ b/foreign/node/src/wire/vsr/index.ts @@@ -21,7 -21,7 +21,6 @@@ import type { CommandResponse } from '. import { COMMAND_CODE } from '../command.code.js'; import { responseError } from '../error.utils.js'; import { HEADER_SIZE, encodeRequestHeader } from './header.js'; --import { namespaceForRequest } from './namespace.js'; import { Operation, isPartition, @@@ -66,7 -66,7 +65,6 @@@ export class VsrSession const operation = registerCommand(command) ? Operation.Register : operationForCode(command); -- const namespace = namespaceForRequest(command, payload, operation); const size = HEADER_SIZE + payload.length; if (size > MAX_U32) throw new RangeError('VSR request exceeds the u32 frame-size limit'); @@@ -94,7 -94,7 +92,6 @@@ client: this.state.clientId, request, operation, -- namespace, session, nonReplicatedCode: operation === Operation.NonReplicated ? command : undefined, diff --cc foreign/node/src/wire/vsr/namespace.test.ts index 244ca3932,244ca3932..000000000 deleted file mode 100644,100644 --- a/foreign/node/src/wire/vsr/namespace.test.ts +++ /dev/null @@@ -1,342 -1,342 +1,0 @@@ --// Licensed to the Apache Software Foundation (ASF) under one --// or more contributor license agreements. See the NOTICE file --// distributed with this work for additional information --// regarding copyright ownership. The ASF licenses this file --// to you under the Apache License, Version 2.0 (the --// "License"); you may not use this file except in compliance --// with the License. You may obtain a copy of the License at --// --// http://www.apache.org/licenses/LICENSE-2.0 --// --// Unless required by applicable law or agreed to in writing, --// software distributed under the License is distributed on an --// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY --// KIND, either express or implied. See the License for the --// specific language governing permissions and limitations --// under the License. -- --import assert from 'node:assert/strict'; --import { describe, it } from 'node:test'; --import { serializeIdentifier } from '../identifier.utils.js'; --import { serializeSendMessages } from '../message/message.utils.js'; --import { Partitioning } from '../message/partitioning.utils.js'; --import { Consumer, serializeStoreOffset } from '../offset/offset.utils.js'; --import { COMMAND_CODE } from '../command.code.js'; --import { ResponseError } from '../error.utils.js'; --import { -- METADATA_CONSENSUS_NAMESPACE, -- namespaceForRequest, -- packNamespace --} from './namespace.js'; --import { Operation } from './operation.js'; -- --describe('VSR namespace routing', () => { -- it('routes register and logout to metadata consensus', () => { -- for (const operation of [Operation.Register, Operation.Logout]) -- assert.equal( -- namespaceForRequest(0, Buffer.alloc(0), operation), -- METADATA_CONSENSUS_NAMESPACE -- ); -- }); -- -- it('routes metadata and non-replicated requests to zero', () => { -- assert.equal( -- namespaceForRequest( -- COMMAND_CODE.CreateStream, -- Buffer.alloc(0), -- Operation.CreateStream -- ), -- 0n -- ); -- assert.equal( -- namespaceForRequest( -- COMMAND_CODE.GetStats, -- Buffer.alloc(0), -- Operation.NonReplicated -- ), -- 0n -- ); -- }); -- -- it('packs explicit send-message partition identifiers', () => { -- const payload = serializeSendMessages( -- 7, -- 11, -- [], -- Partitioning.PartitionId(13) -- ); -- assert.equal( -- namespaceForRequest( -- COMMAND_CODE.SendMessages, -- payload, -- Operation.SendMessages -- ), -- packNamespace(7, 11, 13) -- ); -- }); -- -- it('defers named stream and topic routing to the server', () => { -- const payload = serializeSendMessages( -- 'stream', -- 'topic', -- [], -- Partitioning.PartitionId(1) -- ); -- assert.equal( -- namespaceForRequest( -- COMMAND_CODE.SendMessages, -- payload, -- Operation.SendMessages -- ), -- 0n -- ); -- }); -- -- it('rejects server-selected message partitioning', () => { -- const payload = serializeSendMessages( -- 1, -- 2, -- [], -- Partitioning.Balanced -- ); -- assert.throws( -- () => namespaceForRequest( -- COMMAND_CODE.SendMessages, -- payload, -- Operation.SendMessages -- ), -- (error: unknown) => -- error instanceof ResponseError && error.errorCode === 5 -- ); -- }); -- -- it('routes consumer offsets from their explicit partition', () => { -- const payload = serializeStoreOffset( -- 1, -- 2, -- Consumer.Single, -- 3, -- 99n -- ); -- assert.equal( -- namespaceForRequest( -- COMMAND_CODE.StoreOffset, -- payload, -- Operation.StoreConsumerOffset -- ), -- packNamespace(1, 2, 3) -- ); -- }); -- -- it('routes delete-segments payloads', () => { -- const payload = Buffer.concat([ -- serializeIdentifier(1), -- serializeIdentifier(2), -- Buffer.from([3, 0, 0, 0]) -- ]); -- assert.equal( -- namespaceForRequest( -- COMMAND_CODE.DeleteSegments, -- payload, -- Operation.DeleteSegments -- ), -- packNamespace(1, 2, 3) -- ); -- }); -- -- it('accepts the maximum packable identifiers', () => { -- assert.equal( -- packNamespace(4095, 4095, 999_999), -- (4095n << 32n) | (4095n << 20n) | 999_999n -- ); -- }); -- -- it('rejects peeks past the declared send-messages metadata region', () => { -- const payload = serializeSendMessages( -- 1, -- 2, -- [], -- Partitioning.PartitionId(3) -- ); -- const underDeclared = Buffer.from(payload); -- // Shrink the declared metadata region so the partitioning bytes sit -- // outside it; the peek must fail instead of reading them. -- underDeclared.writeUInt32LE(payload.readUInt32LE(0) - 6, 0); -- assert.throws( -- () => namespaceForRequest( -- COMMAND_CODE.SendMessages, -- underDeclared, -- Operation.SendMessages -- ), -- (error: unknown) => -- error instanceof ResponseError && error.errorCode === 3 -- ); -- }); -- -- it('rejects unknown codes in partition routing', () => { -- assert.throws( -- () => namespaceForRequest( -- 60_001, -- Buffer.alloc(0), -- Operation.SendMessages -- ), -- (error: unknown) => -- error instanceof ResponseError && error.errorCode === 5 -- ); -- }); -- -- it('requires an explicit consumer-offset partition', () => { -- const payload = serializeStoreOffset( -- 1, -- 2, -- Consumer.Single, -- 3, -- 99n -- ); -- // [kind u8][consumer 6][stream 6][topic 6] puts the partition flag at 19. -- const withoutPartition = Buffer.from(payload); -- withoutPartition.writeUInt8(0, 19); -- assert.throws( -- () => namespaceForRequest( -- COMMAND_CODE.StoreOffset, -- withoutPartition, -- Operation.StoreConsumerOffset -- ), -- (error: unknown) => -- error instanceof ResponseError && error.errorCode === 6 -- ); -- }); -- -- it('rejects namespace fields before masking', () => { -- for (const [streamId, topicId, partitionId] of [ -- [4096, 0, 0], -- [0, 4096, 0], -- [0, 0, 1_000_000] -- ] as const) -- assert.throws( -- () => packNamespace(streamId, topicId, partitionId), -- (error: unknown) => -- error instanceof ResponseError && error.errorCode === 6 -- ); -- }); -- -- it('rejects negative and non-integer namespace fields', () => { -- for (const [streamId, topicId, partitionId] of [ -- [-1, 0, 0], -- [0, -1, 0], -- [0, 0, -1], -- [0.5, 0, 0], -- [0, Number.NaN, 0] -- ] as const) -- assert.throws( -- () => packNamespace(streamId, topicId, partitionId), -- (error: unknown) => -- error instanceof ResponseError && error.errorCode === 6 -- ); -- }); -- -- it('rejects malformed identifiers at every prefix boundary', () => { -- const payload = serializeSendMessages( -- 1, -- 2, -- [], -- Partitioning.PartitionId(3) -- ); -- for (let length = 0; length < payload.length; length += 1) -- assert.throws( -- () => namespaceForRequest( -- COMMAND_CODE.SendMessages, -- payload.subarray(0, length), -- Operation.SendMessages -- ), -- ResponseError -- ); -- -- const invalidKind = Buffer.from(payload); -- invalidKind.writeUInt8(99, 4); -- assert.throws( -- () => namespaceForRequest( -- COMMAND_CODE.SendMessages, -- invalidKind, -- Operation.SendMessages -- ), -- ResponseError -- ); -- }); -- -- it('rejects malformed consumer-offset and delete-segment payloads', () => { -- const offsetPayload = serializeStoreOffset( -- 1, -- 2, -- Consumer.Single, -- 3, -- 99n -- ); -- const invalidConsumerKind = Buffer.from(offsetPayload); -- invalidConsumerKind.writeUInt8(0, 0); -- assert.throws( -- () => namespaceForRequest( -- COMMAND_CODE.StoreOffset, -- invalidConsumerKind, -- Operation.StoreConsumerOffset -- ), -- ResponseError -- ); -- for (let length = 1; length < 20; length += 1) -- assert.throws( -- () => namespaceForRequest( -- COMMAND_CODE.StoreOffset, -- offsetPayload.subarray(0, length), -- Operation.StoreConsumerOffset -- ), -- ResponseError -- ); -- -- const deletePayload = Buffer.concat([ -- serializeIdentifier(1), -- serializeIdentifier(2), -- Buffer.from([3, 0, 0, 0]) -- ]); -- for (let length = 0; length < deletePayload.length; length += 1) -- assert.throws( -- () => namespaceForRequest( -- COMMAND_CODE.DeleteSegments, -- deletePayload.subarray(0, length), -- Operation.DeleteSegments -- ), -- ResponseError -- ); -- }); -- -- it('defers named offset and delete-segment routing to the server', () => { -- const offsetPayload = serializeStoreOffset( -- 'stream', -- 'topic', -- Consumer.Single, -- 3, -- 99n -- ); -- assert.equal( -- namespaceForRequest( -- COMMAND_CODE.StoreOffset, -- offsetPayload, -- Operation.StoreConsumerOffset -- ), -- 0n -- ); -- -- const deletePayload = Buffer.concat([ -- serializeIdentifier('stream'), -- serializeIdentifier('topic'), -- Buffer.from([3, 0, 0, 0]) -- ]); -- assert.equal( -- namespaceForRequest( -- COMMAND_CODE.DeleteSegments, -- deletePayload, -- Operation.DeleteSegments -- ), -- 0n -- ); -- }); --}); diff --cc foreign/node/src/wire/vsr/namespace.ts index 1a6ec20e9,1a6ec20e9..000000000 deleted file mode 100644,100644 --- a/foreign/node/src/wire/vsr/namespace.ts +++ /dev/null @@@ -1,216 -1,216 +1,0 @@@ --// Licensed to the Apache Software Foundation (ASF) under one --// or more contributor license agreements. See the NOTICE file --// distributed with this work for additional information --// regarding copyright ownership. The ASF licenses this file --// to you under the Apache License, Version 2.0 (the --// "License"); you may not use this file except in compliance --// with the License. You may obtain a copy of the License at --// --// http://www.apache.org/licenses/LICENSE-2.0 --// --// Unless required by applicable law or agreed to in writing, --// software distributed under the License is distributed on an --// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY --// KIND, either express or implied. See the License for the --// specific language governing permissions and limitations --// under the License. --// -- --/** -- * Namespace packing and the partition-plane payload peeks needed to derive -- * it, ported from `core/binary_protocol/src/namespace.rs` and -- * `namespace_for_request` in `core/sdk/src/vsr.rs`. -- */ -- --import { COMMAND_CODE } from '../command.code.js'; --import { responseError } from '../error.utils.js'; --import { Operation, isMetadata } from './operation.js'; -- --/** `IggyError::InvalidCommand`. */ --const INVALID_COMMAND = 3; --/** `IggyError::FeatureUnavailable`. */ --const FEATURE_UNAVAILABLE = 5; --/** `IggyError::InvalidIdentifier`. */ --const INVALID_IDENTIFIER = 6; -- --const MAX_STREAMS = 4096; --const MAX_TOPICS = 4096; --const MAX_PARTITIONS = 1_000_000; --const bitsRequired = (value: number): bigint => -- BigInt(BigInt(value).toString(2).length); --const TOPIC_SHIFT = bitsRequired(MAX_PARTITIONS - 1); --const STREAM_SHIFT = TOPIC_SHIFT + bitsRequired(MAX_TOPICS - 1); -- --/** -- * Control-plane requests target the metadata replica (shard 0), selected by -- * this exact sentinel. Plain 0 would fall into namespace hashing and land a -- * Register on a peer shard. -- */ --export const METADATA_CONSENSUS_NAMESPACE = 1n << 63n; -- --/** Packs stream / topic / partition ids into a routing namespace. */ --export const packNamespace = ( -- streamId: number, -- topicId: number, -- partitionId: number --): bigint => { -- validateField(streamId, MAX_STREAMS); -- validateField(topicId, MAX_TOPICS); -- validateField(partitionId, MAX_PARTITIONS); -- return (BigInt(streamId) << STREAM_SHIFT) | -- (BigInt(topicId) << TOPIC_SHIFT) | -- BigInt(partitionId); --}; -- --/** -- * Selects the routing namespace for a request. Partition-plane commands -- * derive it from their own payload; a named stream or topic identifier -- * yields 0 so the server resolves the name. -- * -- * @throws Error mirroring the Rust SDK: invalid-identifier for an -- * out-of-range field, invalid-command for an undecodable payload, and -- * feature-unavailable for a partition operation this SDK cannot derive. -- */ --export const namespaceForRequest = ( -- code: number, -- payload: Buffer, -- operation: number --): bigint => { -- if (operation === Operation.Register || operation === Operation.Logout) -- return METADATA_CONSENSUS_NAMESPACE; -- if (operation === Operation.NonReplicated || isMetadata(operation)) -- return 0n; -- -- switch (code) { -- case COMMAND_CODE.SendMessages: -- return namespaceFromSendMessages(payload); -- case COMMAND_CODE.StoreOffset: -- case COMMAND_CODE.DeleteConsumerOffset: -- case COMMAND_CODE.StoreOffset2: -- case COMMAND_CODE.DeleteConsumerOffset2: -- return namespaceFromConsumerOffset(payload); -- case COMMAND_CODE.DeleteSegments: -- return namespaceFromDeleteSegments(payload); -- default: -- // The guard that keeps custom partition operations unreachable. -- throw responseError(code, FEATURE_UNAVAILABLE); -- } --}; -- --/** A decoded identifier: numeric value, or null for a name (server resolves). */ --type PeekedIdentifier = { -- numeric: number | null, -- length: number --}; -- --const IDENTIFIER_KIND_NUMERIC = 1; --const IDENTIFIER_KIND_STRING = 2; -- --const peekIdentifier = (payload: Buffer, offset: number): PeekedIdentifier => { -- if (payload.length < offset + 2) -- throw responseError(0, INVALID_COMMAND); -- const kind = payload.readUInt8(offset); -- const length = payload.readUInt8(offset + 1); -- if (payload.length < offset + 2 + length) -- throw responseError(0, INVALID_COMMAND); -- if (kind === IDENTIFIER_KIND_NUMERIC) { -- if (length !== 4) -- throw responseError(0, INVALID_COMMAND); -- return { numeric: payload.readUInt32LE(offset + 2), length: 2 + length }; -- } -- if (kind === IDENTIFIER_KIND_STRING && length > 0) -- return { numeric: null, length: 2 + length }; -- throw responseError(0, INVALID_COMMAND); --}; -- --const validateField = (value: number, exclusiveMax: number): void => { -- if (!Number.isInteger(value) || value < 0 || value >= exclusiveMax) -- throw responseError(0, INVALID_IDENTIFIER); --}; -- --const namespaceFromIds = ( -- stream: PeekedIdentifier, -- topic: PeekedIdentifier, -- partitionId: number --): bigint => { -- // Named identifiers defer resolution to the server. -- if (stream.numeric === null || topic.numeric === null) return 0n; -- return packNamespace(stream.numeric, topic.numeric, partitionId); --}; -- --/** -- * `SendMessages`: `[metadata_len u32][stream ident][topic ident] -- * [partitioning kind u8, len u8, value]...`. Only explicit `PartitionId` -- * partitioning is routable under VSR; the broker never picks a partition. -- */ --const namespaceFromSendMessages = (payload: Buffer): bigint => { -- if (payload.length < 4) -- throw responseError(COMMAND_CODE.SendMessages, INVALID_COMMAND); -- const metadataLength = payload.readUInt32LE(0); -- if (payload.length < 4 + metadataLength) -- throw responseError(COMMAND_CODE.SendMessages, INVALID_COMMAND); -- // Rust peeks inside payload[4..4 + metadata_length]; a read past the -- // declared metadata region must fail rather than spill into message bytes -- // and derive a namespace the server would never compute. -- const metadata = payload.subarray(4, 4 + metadataLength); -- -- let offset = 0; -- const stream = peekIdentifier(metadata, offset); -- offset += stream.length; -- const topic = peekIdentifier(metadata, offset); -- offset += topic.length; -- -- if (metadata.length < offset + 2) -- throw responseError(COMMAND_CODE.SendMessages, INVALID_COMMAND); -- const partitioningKind = metadata.readUInt8(offset); -- const partitioningLength = metadata.readUInt8(offset + 1); -- const PARTITIONING_PARTITION_ID = 2; -- if (partitioningKind !== PARTITIONING_PARTITION_ID) -- throw responseError(COMMAND_CODE.SendMessages, FEATURE_UNAVAILABLE); -- if (partitioningLength !== 4 || metadata.length < offset + 2 + 4) -- throw responseError(COMMAND_CODE.SendMessages, INVALID_COMMAND); -- const partitionId = metadata.readUInt32LE(offset + 2); -- -- return namespaceFromIds(stream, topic, partitionId); --}; -- --/** -- * Consumer-offset requests: `[consumer kind u8][consumer ident] -- * [stream ident][topic ident][partition flag u8][partition u32]...`. -- */ --const namespaceFromConsumerOffset = (payload: Buffer): bigint => { -- if (payload.length < 1 || (payload.readUInt8(0) !== 1 && -- payload.readUInt8(0) !== 2)) -- throw responseError(COMMAND_CODE.StoreOffset, INVALID_COMMAND); -- let offset = 1; -- const consumer = peekIdentifier(payload, offset); -- offset += consumer.length; -- const stream = peekIdentifier(payload, offset); -- offset += stream.length; -- const topic = peekIdentifier(payload, offset); -- offset += topic.length; -- -- if (payload.length < offset + 5) -- throw responseError(COMMAND_CODE.StoreOffset, INVALID_COMMAND); -- const hasPartition = payload.readUInt8(offset) === 1; -- if (!hasPartition) -- throw responseError(COMMAND_CODE.StoreOffset, INVALID_IDENTIFIER); -- const partitionId = payload.readUInt32LE(offset + 1); -- -- return namespaceFromIds(stream, topic, partitionId); --}; -- --/** `DeleteSegments`: `[stream ident][topic ident][partition u32]...`. */ --const namespaceFromDeleteSegments = (payload: Buffer): bigint => { -- let offset = 0; -- const stream = peekIdentifier(payload, offset); -- offset += stream.length; -- const topic = peekIdentifier(payload, offset); -- offset += topic.length; -- -- if (payload.length < offset + 4) -- throw responseError(COMMAND_CODE.DeleteSegments, INVALID_COMMAND); -- const partitionId = payload.readUInt32LE(offset); -- -- return namespaceFromIds(stream, topic, partitionId); --};
