Act on registration rejections and reclaim stale leases - #9
Conversation
Fixes #8. A registration the relay refuses was routed through `snafu_error_get_or_return_ok!`, which returns `Ok(())` — the success channel — so the worker loop read a refusal as "finished without an error" and reconnected on the transport ladder's 1s cap. A permanent rejection retried forever; a quota rejection retried ~60x a minute. The rejection is now an answer rather than a failure to get one: `Status::Rejected` ends the worker, and `Status::RejectedRetryable` waits on its own ladder (`PB_MAPPER_REGISTRATION_REJECT_BACKOFF_MIN` /`_MAX`, 5s to 80s). The two ladders live in a `ControlBackoff` pair so a reconnect, which happens on its own schedule, can no longer reset the wait a full quota was serving. The relay handed out `lease_ttl_ms` but never enforced it, so a client that neither renewed nor disconnected held its slot until its process died. A ticker task now sends `SweepServerLeases` into the manager channel — from outside the loop, since that loop is the channel's only consumer and its single await is a receive. The sweep marks a registration Suspect on the first pass past the threshold, which drops it from the subscribe candidates, and retires it on the second. Legacy (v1) registrations keep their far longer grace, since the two protocol versions keep a registration alive differently. For a relay already wedged, `pb-mapper admin connection retire <service> --all` frees the quota without a restart, reaching through the SDK, the CLI, and the napi surface. Also deletes what nothing called: four unused config accessors, `TimeoutCount`, `ADMIN_KEY_PATH`, `take_context`, `ServiceStatus`, two manager message helpers, `forwards_now`, a placeholder test, and eight error variants with no generated selector. Tests: `lease_reclaim.rs` reproduces the incident — a registration that reads but never renews is retired, one that keeps pinging is not. `regression.rs` asserts a retryable rejection waits on the slow ladder and a terminal one ends the worker after exactly one attempt. `sdk_e2e.rs` and `js/test/e2e.test.mjs` cover retirement. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 670cdcfe6e
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| let retired = query_inventory( | ||
| authorization, | ||
| &manager, | ||
| ManagerTask::AdminConnectionRetire { | ||
| key, | ||
| conn_id: conn_id.map(RemoteConnId::from), | ||
| response_sender, | ||
| }, | ||
| receiver, | ||
| "connection retire", | ||
| ) |
There was a problem hiding this comment.
If the administrator key rotates after this task reaches the manager but before the oneshot response arrives, query_inventory returns administrator_key_rotated even though AdminConnectionRetire has already removed the connections. The subsequent audit is skipped as well, so the CLI reports failure for an unaudited mutation that actually occurred; retirement needs dispatch semantics that establish its authorization/audit point before applying the side effect rather than reusing the read-only inventory helper.
Useful? React with 👍 / 👎.
| let namespace = key_id.unwrap_or_default(); | ||
| let key = compose_service_key(namespace, &service_name); |
There was a problem hiding this comment.
When a protocol or SDK caller supplies key_id: None and a service name such as @0000000000000001\0foo, compose_service_key uses that string verbatim, so it matches the internal routing key for namespace 1 and retires that service while the audit records an unscoped target. Normal registration explicitly rejects NUL-containing service names to prevent this key ambiguity; apply the same validation to this wire-supplied name before composing the retirement key.
Useful? React with 👍 / 👎.
|
|
||
| #[tokio::test] | ||
| async fn retryable_registration_rejection_waits_on_the_slow_ladder() { | ||
| let _pool_size = EnvVarGuard::set("PB_MAPPER_CONTROL_CONN_POOL_SIZE", "1"); |
There was a problem hiding this comment.
When this integration-test binary runs cases concurrently, both new rejection tests and local_server_reconnects_when_registered_conn_is_missing_from_remote_status mutate PB_MAPPER_CONTROL_CONN_POOL_SIZE with independent guards. One test can restore or remove the variable before another tunnel reads it, causing that tunnel to use the default two-worker pool; the retryable case can then observe two immediate attempts, while the terminal case can observe two attempts instead of one. Protect these overrides with a process-wide lock or isolate the cases in a dedicated test binary.
AGENTS.md reference: AGENTS.md:L52-L53
Useful? React with 👍 / 👎.
| registration | ||
| .wait_ready_timeout(READY_TIMEOUT) | ||
| .await | ||
| .unwrap(); |
There was a problem hiding this comment.
Immediately after the admin response, the registration's watch value may still be Connected from the original control pool, so wait_ready_timeout returns immediately without waiting for any worker to process retirement or register again. This test therefore passes even if retirement permanently leaves the service absent; poll the relay for replacement connection IDs and round-trip a payload through the tunnel before asserting recovery.
AGENTS.md reference: AGENTS.md:L56-L58
Useful? React with 👍 / 👎.
| @@ -176,22 +176,6 @@ pub async fn resolve_addrs_async(addr: &str) -> Result<ResolvedAddrs> { | |||
| ResolvedAddrs::new(addr, addrs, parse_error) | |||
| } | |||
|
|
|||
There was a problem hiding this comment.
The repository marks 0.4.0 as a published release containing the shipped workspace crates, but this cleanup removes public pb-mapper-core functions such as get_sockaddr, get_sockaddr_async, and the two get_pb_mapper_server* helpers while the workspace version remains 0.4.0. External callers that used these exported wrappers will stop compiling on an otherwise same-version update; retain deprecated forwarding wrappers or make the removal as part of an explicit breaking-version release. The same cleanup also removes other exported symbols such as TimeoutCount and ServerHeaderSession::take_context, so the public-API surface should be audited rather than treating repository-internal references as the only users.
Useful? React with 👍 / 👎.
Four review findings on the retirement path and its tests. `ConnectionRetire` reused `query_inventory`, whose authorization point sits *after* the manager answers. For a read that is right — a snapshot taken under a credential that has since rotated should not be handed back. For a mutation it is wrong: by the time the manager answers the connections are already gone, so a root rotation in that window made the CLI report `administrator_key_rotated` for a retirement that happened, and skipped auditing it. Split out `apply_mutation`, which authorizes before sending and reports the outcome unconditionally, with the send half shared as `send_manager_task`. `audit_read` is now `audit_action`, since it records mutations too. The retire service name arrived from the wire and went straight into `compose_service_key`, so a name carrying the NUL separator could spell out another namespace's routing key — retiring a service the audit record did not name. Registration already rejected such names; that check is now `validate_service_name`, called by both paths. In `regression.rs`, three tests overrode the same `PB_MAPPER_CONTROL_CONN_POOL_SIZE` with independent guards. Run concurrently, one test's guard could restore the variable while another's tunnel was still reading it, silently giving that tunnel the default two-worker pool and a second register attempt where the test asserts one. `EnvOverrides` holds a lock for as long as the values are in place. The `sdk_e2e` retire test waited on `wait_ready_timeout`, which returns immediately while the status still reads Connected from the pool that was just retired — so it passed whether or not the service came back. It now waits for replacement conn_ids in the relay's own view and round-trips a payload through the tunnel. Verified by inverting the id set: the test fails without a reconnect. Not changed: the review also asked for deprecated forwarding wrappers around the removed `pb-mapper-core` helpers on semver grounds. These crates have never been published — no `cargo publish` anywhere in the repo, and crates.io has no `pb-mapper*` — so v0.4.0 is a GitHub binary release with no external API consumers to break. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: a390553fde
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| ) { | ||
| let conn_sender = manager.get_conn_sender_chan(&conn_id); | ||
| let removed_from_service_map = remove_server_conn(server_conn_map, key, conn_id); | ||
| let removed_from_active_map = manager.deregister_conn(conn_id); |
There was a problem hiding this comment.
When an admin or lease-sweep retirement races with a newly accepted socket, this deregistration immediately returns conn_id to the idle pool even though the retired handle_server_conn still owns a ServerConnGuard. The Accept arm can reuse that ID and register a replacement before the old guard sends DeRegisterServerConn; that stale task then removes the replacement's active sender and, when it uses the same service key, its service-map entry. Defer recycling until the old socket task has completed, or make deregistration generation-aware.
Useful? React with 👍 / 👎.
| PB_MAPPER_REGISTRATION_REJECT_BACKOFF_MIN, | ||
| DEFAULT_REGISTRATION_REJECT_BACKOFF_MIN, | ||
| ) | ||
| .max(Duration::from_millis(1)); |
There was a problem hiding this comment.
When PB_MAPPER_REGISTRATION_REJECT_BACKOFF_MIN is set to 0s, this converts it to 1 ms instead of falling back to a safe value. Because RetryBackoff caps its multiplier at 1024, the delay then never exceeds roughly one second even when the maximum remains at its default 80 seconds; if both variables are zero, rejected registrations retry every millisecond. This recreates the rejection hammering the new ladder is intended to prevent, so zero should select the default or be rejected rather than treated as a usable minimum.
Useful? React with 👍 / 👎.
| PB_MAPPER_SERVER_LEASE_SWEEP_INTERVAL, | ||
| DEFAULT_SERVER_LEASE_SWEEP_INTERVAL, | ||
| ) | ||
| .max(Duration::from_millis(1)) |
There was a problem hiding this comment.
When PB_MAPPER_SERVER_LEASE_SWEEP_INTERVAL is accidentally configured as 0s, this clamp starts a ticker with a 1 ms period. Every tick queues an O(number of registrations) scan through the single routing-manager loop, so a configuration value intended to be made nonfatal can instead consume CPU and delay normal routing work. Treat zero as invalid and use the five-second default, or enforce a practical lower bound.
Useful? React with 👍 / 👎.
Three findings from the second review round on this PR. Separate dropping a connection's task sender from recycling its ID. `retire_server_conn` did both at once, but the retired socket task is still running at that point and still owns a `ServerConnGuard` that will send `DeRegisterServerConn` when it unwinds. With the ID already back in the idle pool, an `Accept` racing the retirement could hand it to a new socket, let a replacement register under it, and then have the old guard's deregistration — which matches on conn_id alone — remove the replacement's active sender and, on the same service key, its service-map entry. The ID now stays out of circulation until that deregistration arrives, which is the one point where nothing will speak for it again; every registered connection produces exactly one such task, so no ID is stranded. `TaskManager::deregister_conn` is gone in favour of the two explicit steps, since no caller wanted them fused. Treat a zero `PB_MAPPER_REGISTRATION_REJECT_BACKOFF_MIN`/`_MAX` and a zero `PB_MAPPER_SERVER_LEASE_SWEEP_INTERVAL` as absent rather than clamping them to a millisecond. The clamp avoided a start-up panic but traded it for a hot loop: `RetryBackoff` caps its multiplier at 1024, so a millisecond minimum never climbs past about a second however high the maximum is set, recreating the reject hammering the ladder exists to prevent — and a millisecond sweep period queues a scan over every registration into the single routing loop that often. A zero now selects the default and logs why. An inverted backoff range still collapses to a fixed delay, which is a legible request rather than a typo. Verified: fmt, clippy `-D warnings`, `cargo test --workspace --all-features` (199 tests, 22 binaries), `bun run build:release` + `bun test`. The new runtime test was confirmed to fail when recycling is restored to retirement. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 65c2d9c1f8
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| sender | ||
| .try_send(ConnTask::Retire { | ||
| reason: reason.to_string(), | ||
| }) | ||
| .ok() |
There was a problem hiding this comment.
When the control writer is already blocked in handle_stream_req(...).await because the registration has stopped reading, queuing ConnTask::Retire on that same writer's channel cannot be observed until the blocked write returns. The manager nevertheless removes the registration and reports it retired, while its TCP task remains alive, its client never reconnects, and its connection ID cannot be recycled; a half-open registration that still sends inbound traffic can remain in this state indefinitely. Retirement needs to cancel/abort the writer or close the socket independently of its work queue.
Useful? React with 👍 / 👎.
`ConnTask::Retire` reaches the control-connection writer only between writes, and the server puts no timeout on a control write: a registration that has stopped reading leaves `write_msg` blocked indefinitely, so a retirement queued behind it is never observed. The reader-side idle timeout does not bound the task either — it only fires while a read is outstanding, and each Ping renews the lease and resets health to `Healthy`. The manager would drop the registration, log it retired, and leave the socket task alive; since a connection ID is not recycled until that task's guard deregisters, the ID leaked with it. Each registered control connection now carries a `CancellationToken`, stored in `ServerConnInfo` because the service map is the only structure that outlives the socket task's own state. `retire_server_conn` cancels it after sending the graceful notification, so a responsive writer still unwinds through `ConnTask::Retire` and a wedged one is interrupted mid-write. The two subscribe-path sites that drop an unusable registration cancel as well, via `cancel_and_remove_server_conn`. The writer loop moved into `run_control_writer`, which races the loop as a whole against the token rather than adding a `select!` arm — an arm is polled only between iterations, which is exactly the gap being closed. Dropping a write half-way is safe: the borrowed writer is the only handle to the socket's write side and the caller drops it immediately after. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
![claude[bot]](https://avatars.githubusercontent.com/in/1236702?s=60&v=4){kind=small}
| /// Retire every connection the service has. Required to be explicit, | ||
| /// because it is what frees a full quota and also what interrupts every | ||
| /// healthy connection at once. | ||
| #[arg(long, group = "target", default_value_t = false)] |
There was a problem hiding this comment.
all carries default_value_t = false while being a member of the target ArgGroup::new("target").required(true). clap satisfies a group's required check on plain presence in ArgMatches, which a defaulted value already provides — so the group's requirement is trivially satisfied even when the user passes neither --conn-id nor --all.
Concrete failure: running pb-mapper admin connection retire my-service with no target flag does not error as the group's required(true) intends. Instead all resolves to its default false and conn_id to None, so let conn_id = if all { None } else { conn_id }; yields None — identical to explicitly passing --all. The command silently retires every connection for the service, which is exactly the "must be explicit" guard the PR description says this group exists to enforce.
Fix: drop default_value_t = false (a derived bool flag already defaults to false via ArgAction::SetTrue without needing to be a group default), or validate conn_id.is_some() != all explicitly after parsing instead of relying on clap's group requiredness.
Four rounds of review left three kinds of redundancy behind, each solved locally at the time. `retire_server_conn` took seven parameters, four of which were one group of routing state threaded separately through every call site, including a bare three-tuple whose fields were named nowhere. `RoutingState` borrows the four as a unit and two aliases name the map types. The paging block — clamp, skip, take, decide next_page — was written out three times, and its 1000 ceiling four times, once on the client where it has to agree with a server constant it could not see. `pb-mapper-core` is the lowest layer both `auth` and `server` reach, so `paging` lives there and the SDK takes the ceiling from it. Sorting stays at each call site; the three orderings differ and belong to their listings. Registering a v2 control connection was written out in two test files, differing only in instance id, heartbeat, and timeout. It moves to the testkit behind a defaulted spec. Behaviour is unchanged throughout: same log fields, error codes, and wire format. The retirement e2e is new coverage rather than tidying. Unit coverage stopped at the writer returning, so the rest of the chain — socket task unwound, guard deregistered, ID reclaimed — was only ever confirmed by reading. The case registers a connection that pings but never drains, so its lease stays renewed and only retirement can end it, then asserts both that it leaves the admin listing and that this side's writes start failing. Verified by mutation: leaving the socket task alive while removing the routing entry — the incident's own shape — fails the second assertion. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
1 participant
Fixes #8, and deletes the dead code found while going through the tree.
1. A rejection was laundered into a reconnect
PbConnResponse::Errorwent throughsnafu_error_get_or_return_ok!, whichexpands to
return Ok(()). That is the success channel, so the worker loop reada refusal as "finished without an error", took the transport backoff (100ms →
1s cap, unlimited attempts) and looped. The
retryableflag was logged anddiscarded.
A rejection is an answer, not a failure to get one, so it no longer travels that
path:
Status::Rejected(reason)— terminal. The worker stops; the pool returns.Status::RejectedRetryable(reason)— waits on a ladder of its own,PB_MAPPER_REGISTRATION_REJECT_BACKOFF_MIN/_MAX, defaulting to 5s → 80s.Both are clamped when read, so an env typo cannot trip
RetryBackoff::new'sassertions at start-up.
The two ladders sit in a
ControlBackoffpair rather than sharing one counter. Atransport failure and a quota rejection are not the same kind of problem, and
sharing meant a reconnect — which happens on its own schedule — reset the wait a
full quota was serving.
2. The relay never enforced the lease it handed out
lease_ttl_mswent to the client and nothing checked it, so a client thatneither renewed nor disconnected held its slot until its process died.
last_rx_atwas already tracked; only the timer was missing.
A ticker task now sends
ManagerTask::SweepServerLeases. It sends from outsidethe manager loop deliberately: that loop is its channel's only consumer and its
single await is a receive, so racing a tick against the receive in
select!would drop a task the receive had already taken, and sending into its own channel
would have it waiting on itself. The tick is dropped on a full queue.
The sweep is two-stage — Healthy → Suspect on the first pass past the threshold,
which drops the connection from the subscribe candidates, then retired on the
second. The threshold is
server_lease_timeout()for protocol v2 and the farlonger legacy idle timeout for v1, since the two versions keep a registration
alive differently and a sweep stricter than the reader's own timeout would retire
healthy connections.
3. An operator path for a relay that is already wedged
Mutating, so admin-only, protocol-v2-only, and replay-protected like every other
mutation. Reaches through the SDK (
Admin::retire_connections), the CLI, and thenapi surface (
admin.retireConnections()). Zero retired is a normal answer — thetarget may have unwound on its own first. The admin connection listing also
gained an
IDLE MScolumn and thenext_pageline it was missing.4. Dead code
Nothing referenced any of it:
get_sockaddr,get_sockaddr_async,get_pb_mapper_server,get_pb_mapper_server_async,ADMIN_KEY_PATH,TimeoutCount,take_context,ServiceStatus,active_conn_id_msg,idle_conn_id_msg,forwards_now, atest_sleepplaceholder test, and eighterror variants with no generated
*Snafuselector.ServerConnHealth::Suspectstayed — the new sweep constructs it.
Testing
crates/pb-mapper-cli/tests/lease_reclaim.rs(new) reproduces the incident:a registration that keeps reading but never renews its lease is retired, and
one that keeps pinging is left alone. Its own test binary, because it shortens
the process-global lease timeout.
crates/pb-mapper-cli/tests/regression.rs: a fake relay that refuses everyregistration, asserting a retryable rejection waits past the reject ladder's
minimum (so it was not retried as a transport failure) and a terminal one ends
the worker after exactly one attempt.
crates/pb-mapper-server/src/lib.rs: five unit tests over the sweep decision —renewing, mark-then-retire, activity clearing suspicion, the legacy grace, and
every stale connection of a service being reported.
crates/pb-mapper-cli/tests/sdk_e2e.rsandjs/test/e2e.test.mjs: retirementend to end, including retiring one connection by id out of a pool.
cargo fmt --all -- --check,cargo clippy --workspace --all-targets --all-features -- -D warnings, andcargo test --workspace --all-featuresareclean;
bun run build:releaseplusbun testpass, with the addon at 2.07 MiBagainst CI's 5.5 MiB gate.
🤖 Generated with Claude Code