github-actions[bot] commented on code in PR #4387:
URL: https://github.com/apache/iggy/pull/4387#discussion_r4173794166
##########
core/server/src/http/submit.rs:
##########
@@ -406,75 +410,62 @@ pub(in crate::http) async fn partition_write_replicated(
);
PartitionWriteError::Unavailable
})?;
- dispatch_partition_request(
- &state.shard,
- message,
- session.client_id,
- session.session,
- session.client_id,
- Some(session.user_id),
- None,
- )
- .await;
- drop(next_data_request_id);
- let outcome = compio::time::timeout(PARTITION_WRITE_REPLY_TIMEOUT,
receiver).await;
- // Removes the slot unless the reply already fired, so a late commit
- // reply after a timeout sheds at the bus instead of leaking a waiter.
+ let attachment = session.attachment.borrow().clone();
+ let outcome =
+
compio::time::timeout(deadline.saturating_duration_since(Instant::now()), async
{
+ dispatch_partition_request(
+ &state.shard,
+ message,
+ session.client_id,
+ session.session,
+ session.client_id,
+ Some(session.user_id),
+ Some((session.client_id, attachment)),
+ )
+ .await;
+ if let Some(dispatched) = dispatched.take() {
+ let _ = dispatched.send(Ok(()));
+ }
+ receiver.await
+ })
+ .await;
drop(guard);
match outcome {
Ok(Ok(reply)) => {
let reply = reply.into_contiguous();
classify_partition_reply(&reply).map(|header| (reply, header))
}
- // Cancelled (reply target torn down by session eviction mid-wait) or
- // elapsed: same caller contract either way - outcome unknown, 504.
Ok(Err(_)) | Err(_) => Err(PartitionWriteError::Timeout(operation)),
}
}
-/// Fire-and-forget produce (`?ack=none`): installs no reply slot and never
-/// awaits a commit. The commit still happens; its reply (and any gate-failure
-/// reply) targets a request id with no slot installed and is shed at the bus
by
-/// design.
-///
-/// Admitted against the in-flight caps exactly like the acked path
-/// ([`partition_write_replicated`]): being reply-less does not make it free.
-/// The request still parks inside [`dispatch_partition_request`] for the
-/// routable-wait budget while pinning its buffered body, so it must count
-/// against the per-session and shard-global budgets or it would be an uncapped
-/// bypass. The in-flight guard is held across dispatch and released by `Drop`.
-/// A refusal is a synchronous 429/503, which is honest for `?ack=none`:
-/// admission runs before dispatch, so it is not a commit reply the caller
-/// opted out of - it is the request being turned away up front.
+/// Return after dispatch while a detached task holds the partition gate until
+/// the reply or deadline. A 202 still carries no commit or admission
guarantee.
pub(in crate::http) async fn produce_unacked(
- state: &HttpInner,
- session: &HttpSession,
- body: &[u8],
+ state: &Rc<HttpInner>,
+ session: &Rc<HttpSession>,
+ body: Bytes,
) -> Result<(), PartitionWriteError> {
- let _in_flight = admit_partition_write(&session.in_flight_writes,
&state.in_flight_writes)?;
- // Same gate as the acked path, for the same ordering reason.
- let mut next_data_request_id = session.data_gate.lock().await;
- let request_id = *next_data_request_id;
- *next_data_request_id += 1;
- let message = build_request_message(
- Operation::SendMessages,
- session.client_id,
- session.session,
- request_id,
- body,
- );
- dispatch_partition_request(
- &state.shard,
- message,
- session.client_id,
- session.session,
- session.client_id,
- Some(session.user_id),
- None,
- )
- .await;
- drop(next_data_request_id);
- Ok(())
+ let state = Rc::clone(state);
+ let session = Rc::clone(session);
+ let (sent, dispatched) = oneshot::channel();
+ compio::runtime::spawn(async move {
Review Comment:
warning: `produce_unacked()` now runs every `?ack=none` produce through
`partition_write()`, so the detached task holds an in-flight permit and a reply
slot until the commit reply or the `504` deadline. Return after dispatch on the
unacked path, as the previous body did.
##########
core/server/src/http/submit.rs:
##########
@@ -343,49 +340,56 @@ pub(in crate::http) async fn logout_session(state:
&HttpInner, session: &Rc<Http
Ok(Ok(_)) => {}
Ok(Err(error)) => warn!(
?error,
- "server HTTP: VSR Logout submit failed; slot lingers until
eviction"
+ "server HTTP: VSR Logout submit failed; session remains until
lease expiry"
),
Err(_canceled) => warn!(
- "server HTTP: VSR Logout task dropped before replying; slot
lingers until eviction"
+ "server HTTP: VSR Logout task dropped before replying; session
remains until lease expiry"
),
}
state.forget_session(session);
}
-/// Run one awaited partition write (produce / consumer-offset write) end to
-/// end: admit against the in-flight caps, install the reply slot, dispatch
-/// into the partition plane, and wait (bounded) for the committed reply.
-///
-/// Slot-before-dispatch is load-bearing: every pre-dispatch gate failure
-/// inside [`dispatch_partition_request`] replies through `send_to_client`,
-/// which fires an installed slot, so one slot catches every exit. The slot
-/// guard borrows the registry, which is why this whole future runs inside
-/// the caller's `SendWrapper` on shard 0.
-///
-/// Hands back the graded reply frame and the header grading read, so a caller
-/// that renders a committed payload can bound the body without parsing the
-/// header again; the offset writes answer 204 and drop both.
pub(in crate::http) async fn partition_write_replicated(
state: &HttpInner,
session: &HttpSession,
operation: Operation,
body: &[u8],
) -> Result<(Frozen<MESSAGE_ALIGN>, ReplyHeader), PartitionWriteError> {
- // Admission sits here rather than before body decode: axum's extractors
- // already buffered and deserialized the body (bounded by the router-wide
- // `DefaultBodyLimit`) before the handler ran, so the caps gate what is
- // actually unbounded - the slot install, the dispatch, and the parked
- // reply await that pins this request's buffers for up to the reply
- // timeout. Held across every exit below; released by `Drop`.
+ partition_write(state, session, operation, body, &mut None).await
+}
+
+async fn partition_write(
+ state: &HttpInner,
+ session: &HttpSession,
+ operation: Operation,
+ body: &[u8],
+ dispatched: &mut Option<oneshot::Sender<Result<(), PartitionWriteError>>>,
+) -> Result<(Frozen<MESSAGE_ALIGN>, ReplyHeader), PartitionWriteError> {
+ let deadline = Instant::now() + PARTITION_WRITE_REPLY_TIMEOUT;
let _in_flight = admit_partition_write(&session.in_flight_writes,
&state.in_flight_writes)?;
+ let namespace = crate::namespace::resolve_partition_request_namespace(
+ &state.shard,
+ operation,
+ body,
+ session.client_id,
+ )
+ .map_err(PartitionWriteError::Rejected)?;
+ let gate = session.partition_gate(namespace);
+ let _partition_guard = compio::time::timeout(
Review Comment:
warning: `partition_write()` holds the new per-namespace gate through the
commit reply, so one credential cannot keep two writes to the same partition in
flight. Release the gate before the reply wait, or state why the ordering needs
the full round trip.
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -6226,7 +6437,45 @@ where
// repair ring - and not even there on a single replica, which keeps no
// ring at all. A miss degrades to a successful send carrying no
// confirmation, a legal answer no client can tell from a real one.
- let committed_batch_stats =
self.resolve_committed_visible_offsets(&drained);
+ let committed_batch_stats =
self.resolve_committed_batch_stats(&drained);
+ let retirements: Vec<_> = drained
Review Comment:
warning: The commit walk builds the `retirements` vector for every drained
entry on every batch, though only `RetireSession` entries use it. Build the
vector only when the batch holds a `RetireSession`.
##########
foreign/node/src/client/client.socket.ts:
##########
@@ -976,8 +987,8 @@ export class CommandResponseStream extends EventEmitter {
if (error instanceof LeaderMovedError)
throw error;
// Once bytes were handed to the socket, a local transport or decode
- // failure leaves the request outcome ambiguous. Register a fresh session
- // rather than replaying that request under a different client identity.
+ // failure leaves the request outcome ambiguous. Drop the transport while
+ // retaining the logical session and its bind proof.
Review Comment:
warning: The Node README still says that a disconnect or eviction
invalidates the session and that later work must register a new session, but
this path retains the session and its bind proof. Update the session text in
`foreign/node/README.md`.
##########
foreign/go/internal/command/code.go:
##########
@@ -25,7 +25,7 @@ const (
GetSnapshotFileCode Code = 11
GetClusterMetadataCode Code = 12
DescribeOptionsCode Code = 13
- AttachConsumerSessionCode Code = 14
+ BindSessionCode Code = 15
Review Comment:
nit: `foreign/go/README.md` still lists binary command 14 for
consumer-session attachment. Change it to 15, the new `BindSessionCode`.
##########
foreign/go/client/tcp/tcp_poll_routing.go:
##########
@@ -288,7 +290,15 @@ func (c *IggyTcpClient) pollOnRoute(ctx context.Context,
key string, payload []b
}
if !slot.attached || slot.parent.client != route.parent.client ||
slot.parent.session != route.parent.session ||
slot.parent.watermark < route.parent.watermark {
- if _, state, err := slot.client.sendPollRequest(exchangeCtx,
uint32(command.AttachConsumerSessionCode), route.parent.bytes()); err != nil {
+ current := c.pollSession.Load()
+ if current == nil || current.parent.client !=
route.parent.client || current.parent.session != route.parent.session {
+ return nil, ierror.ErrTransientNotAccepted
+ }
+ binding, err := vsr.SerializeBindSession(route.parent.bytes(),
iggcon.Version, current.bindSecret)
+ if err != nil {
+ return nil, err
+ }
+ if _, state, err := slot.client.sendPollRequest(exchangeCtx,
uint32(command.BindSessionCode), binding); err != nil {
Review Comment:
nit: The Go poll path discards the `BindSession` reply, while the Rust SDK
rejects a mismatched bound session. Compare the reply session with
`route.parent.session` and fail the route on mismatch.
##########
core/consensus/src/impls.rs:
##########
@@ -419,6 +431,47 @@ where
}
}
+impl<B, P> VsrConsensus<B, P>
+where
+ B: MessageBus,
+ P: Pipeline<Entry = PipelineEntry>,
+{
+ #[must_use]
+ pub fn retry_capacity(&self, table: &crate::ClientTable) -> usize {
+ if table.capacity_committed() {
+ return table.capacity();
+ }
+ self.pipeline
+ .borrow()
+ .head()
+ .filter(|entry| entry.header.retry_capacity != 0)
+ .map_or_else(
+ || table.capacity(),
+ |entry| entry.header.retry_capacity as usize,
+ )
+ }
+
+ /// Both queues own reservations. Clearing or refusing an entry releases it
+ /// automatically, and a commit replaces its reservation with live
protection.
+ #[must_use]
+ pub fn has_retry_capacity(&self, table: &crate::ClientTable, client: u128)
-> bool {
+ if table.contains(client) ||
message_bus::is_auto_commit_client(client) {
+ return true;
+ }
+ let capacity = self.retry_capacity(table);
+ let pipeline = self.pipeline.borrow();
+ let pending: HashSet<_> = pipeline
Review Comment:
warning: `has_retry_capacity()` collects every pending client id into a new
`HashSet` on each admission, and a client without a committed entry pays that
build on every request. Track the pending distinct count in the pipeline
instead.
##########
foreign/node/scripts/check-vsr-protocol.mjs:
##########
@@ -283,16 +295,16 @@ for (const [name, value] of rustOperations) {
}
const protocolVersion =
- rustProtocolCargo.match(/^version = "([0-9]+)\.([0-9]+)\.([0-9]+)/m);
-assert.ok(protocolVersion, 'binary protocol crate version is missing');
+ rustProtocolVersion.match(/pub const IGGY_PROTOCOL_VERSION: u32 =
pack_protocol_version\(([0-9]+), ([0-9]+), ([0-9]+)\);/);
+assert.ok(protocolVersion, 'binary protocol wire version is missing');
const nodePackedVersion = nodeRegister.match(
/export const IGGY_PROTOCOL_VERSION =\s*\(([0-9]+) << 20\) \| \(([0-9]+) <<
10\) \| ([0-9]+);/
);
assert.ok(nodePackedVersion, 'Node packed protocol version source changed');
assert.deepEqual(
[Number(nodePackedVersion[1]), Number(nodePackedVersion[2])],
[Number(protocolVersion[1]), Number(protocolVersion[2])],
- 'Node protocol major.minor differs from iggy_binary_protocol'
+ 'Node protocol major.minor differs from the Rust wire version'
Review Comment:
warning: The drift check compares only the major and minor fields, so a
patch mismatch between the Node mirror and the Rust wire version passes.
Compare `nodePackedVersion[3]` with the patch field too.
##########
core/common/src/traits/binary_transport.rs:
##########
@@ -75,6 +79,34 @@ mod vsr_session_sealed {
/// mutation corrupts the dedup counter or silently breaks at-most-once.
#[async_trait]
pub trait VsrSessionControl: vsr_session_sealed::Sealed + BinaryTransport {
+ async fn session_bind_secret(&self) -> Result<BindSecret, IggyError>;
+ async fn session_identity(&self) -> Result<SessionIdentity, IggyError>;
+ async fn resume_vsr_session(&self) -> Result<(), IggyError>
+ where
+ Self: Sync,
+ {
+ let identity = self.session_identity().await?;
+ let request = BindSessionRequest {
+ version_info: crate::rust_sdk_version_info(self.sdk_version())?,
+ identity,
+ bind_secret: self.session_bind_secret().await?,
+ };
+ let response = self
+ .send_raw_with_response(
+ iggy_binary_protocol::codes::BIND_SESSION_CODE,
+ request.to_bytes(),
+ )
+ .await?;
+ let bound =
+ LoginRegisterResponse::decode_from(&response).map_err(|_|
IggyError::InvalidFormat)?;
+ if bound.session != identity.session {
+ return Err(IggyError::InvalidSession(bound.session));
Review Comment:
warning: This branch returns `IggyError::InvalidSession`, whose message says
the value must be non-zero, though the real conflict is a different bound
session. Return `IggyError::SessionMismatch(identity.session, bound.session)`
as `poll_routing.rs` does.
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -6226,7 +6437,45 @@ where
// repair ring - and not even there on a single replica, which keeps no
// ring at all. A miss degrades to a successful send carrying no
// confirmation, a legal answer no client can tell from a real one.
- let committed_batch_stats =
self.resolve_committed_visible_offsets(&drained);
+ let committed_batch_stats =
self.resolve_committed_batch_stats(&drained);
+ let retirements: Vec<_> = drained
+ .iter()
+ .map(|entry| {
+ if entry.header.operation != Operation::RetireSession {
+ return None;
+ }
+ let prepare =
self.log.journal().inner.repair_entry(entry.header.op)?;
+ SessionIdentity::decode_from(
+ prepare
+ .as_slice()
+ .get(iggy_binary_protocol::HEADER_SIZE..)?,
+ )
+ .ok()
+ })
+ .collect();
+ for ((entry, batch_stats), retirement) in
+ drained.iter().zip(&committed_batch_stats).zip(&retirements)
+ {
+ if self
+ .dedup
+ .commit_capacity(entry.header.retry_capacity as usize)
+ .is_err()
+ || (entry.header.operation == Operation::SendMessages &&
batch_stats.is_none())
Review Comment:
warning: The comment above says a batch-stats miss degrades to a successful
send, but this check turns a `SendMessages` miss into `FatalCommit` and stops
the node. Align the comment with the check, or drop the check.
##########
foreign/node/src/wire/command.code.ts:
##########
@@ -23,7 +23,7 @@ export const COMMAND_CODE = {
GetSnapshot: 11, // @TODO GET_SNAPSHOT_FILE_CODE: u32 = 11
GetClusterMetadata: 12,
DescribeOptions: 13,
- AttachConsumerSession: 14,
+ BindSession: 15,
Review Comment:
nit: `foreign/node/README.md` still lists binary command 14 for
consumer-session attachment. Change it to 15, the new `BindSession` code.
##########
foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/vsr/VsrLoginCodec.java:
##########
@@ -43,7 +43,7 @@ final class VsrLoginCodec {
* Keep in sync with {@code core/binary_protocol/Cargo.toml}; the server
* accepts any client whose major.minor is not newer than its own.
*/
- static final int PROTOCOL_VERSION = (11 << 10); // 0.11.0
+ static final int PROTOCOL_VERSION = (11 << 10) | 1; // 0.11.1
Review Comment:
nit: The javadoc for `PROTOCOL_VERSION` tells readers to keep the value in
sync with `core/binary_protocol/Cargo.toml`, but the wire version now lives in
`core/binary_protocol/src/version.rs`, independent of the crate release. Point
the javadoc at that file.
##########
core/journal/src/partition_journal/segments.rs:
##########
@@ -143,6 +143,53 @@ impl<S: DurableStorage> PartitionPrepareJournal<S> {
}
}
+ /// Move the physical append point over a reserved offset gap.
+ /// The complete preceding chain must already be durably checkpointed.
+ ///
+ /// # Errors
+ /// Returns an error for an uncheckpointed tail, a regressing offset, or a
failed barrier.
+ pub async fn reanchor_segment_storage(&mut self, next_offset: u64) ->
io::Result<()> {
+ self.ensure_healthy()?;
+ let mut segments = self
+ .state
+ .segment_storage
+ .ok_or_else(|| invalid("segment storage is not enabled"))?;
+ if self.state.checkpoint != self.state.head || self.durable_head !=
self.state.head {
+ return Err(invalid("cannot reanchor an uncheckpointed segment
tail"));
+ }
+ if next_offset < segments.tail.position.next_offset {
+ return Err(invalid(
+ "reserved segment offset regresses the physical tail",
+ ));
+ }
+ if next_offset == segments.tail.position.next_offset {
+ return Ok(());
+ }
+ segments.reset_position(SegmentPosition {
+ start_offset: next_offset,
+ length: 0,
+ next_offset,
+ })?;
+ let state = JournalState {
+ segment_storage: Some(segments),
+ ..self.state
+ };
+ self.poisoned = true;
+ self.open_segment_file(
+ segments.tail.generation,
+ next_offset,
+ 0,
+ self.preallocate_segments.then_some(segments.max_size),
+ )
+ .await?;
+ self.sync_segment_files().await?;
Review Comment:
simplification: `open_segment_file()` already syncs these files and clears
both dirty flags, and the fresh generation from `reset_position()` makes its
early return impossible, so this call syncs nothing. Delete the redundant
`sync_segment_files()` call.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
