Implement connect-v2. Fixes #3884

Implements the router-side Connect-V2 sessionless dial path. Dials are authorized locally via the RouterDataModel instead of a controller-issued service session token; circuit creation flows through the existing `CreateCircuitV3` controller endpoint (#3721). Builds on the sdk-golang v2 migration. - Adds `processConnectV2` on `edgeClientConn`: resolves the service by id or name via the RouterDataModel, checks dial access, and dispatches to the controller via `sendCreateCircuitV3Msg`. Supports both `xgEdgeForwarder` (SDK xgress) and `nonXgConnectHandler` flow-control modes, selected by the SDK's `UseXgressToSdkHeader`. - Makes `CircuitId` optional in `DecodeCreateCircuitV3Request`. The V2 router path does not pre-assign a circuit ID; the controller generates it as V1/V2 already do. Without this the decoder rejected the empty header and every V2 dial hung until timeout. Adds a regression test. - Splits `checkAccess` to close a posture-check bypass on the V2 path. The old single `checkAccess` short-circuited to nil for non-OIDC sessions (V1 ran posture at the controller during `CreateSession`); V2 has no such step, so posture would have been skipped. `checkAccess` now always runs the RDM `HasAccess` (policy + posture) check; `checkAccessIfOidc` keeps the OIDC-only gate for the V1 and bind paths. - Sends the V2 `state_connected` on the default (data) sender rather than the control sender. On multi-underlay channels the two senders are independently ordered, so an early terminator payload on the data sender could beat `state_connected` to the SDK and be dropped (channel/v5 has no message-priority API). - Updates `xgEdgeForwarder.lastRx` on every forward path, including the fast `timeout == 0` `TrySend` branch used for normal payload dispatch. The old code only updated it on the `timeout > 0` path, so active V2 circuits looked idle and could be unrouted prematurely. - Adds `state.ConnState.ServiceId`, populated by the connect handlers from the service session token (V1) or the request header (V2). The non-xgress V2 path previously left this empty, so `handleDialAccessLost` could not identify and close V2 non-xgress circuits when dial access was revoked. - Skips conns with no `ServiceSessionToken` in `RemoveLegacyServiceSession`; a sessionless V2 conn's token is nil and the cleanup loop previously dereferenced it unconditionally, which would panic the router. - Advertises Connect-V2 via the `RouterCapabilityConnectV2` bit in the listener hello so SDKs can detect V2 support. - Wires `ContentTypeConnectV2` and `ContentTypeXgControl` handlers in `Acceptor.BindChannel`, and adds `handleXgControl` for SDK-side xgress control messages, preserving `ControlUserVal` so trace-route responses correlate back to the initiator's `SendForReply` waiter. - Adds `RouterDataModel.serviceNameIndex` for O(1) name->id lookup in the V2 dial path, maintained with rename safety at the `HandleServiceEvent` mutation points. - Adds `tests/connect_v2_test.go` covering end-to-end V2 dataflow and the V1 fallback (`ForceConnectV1`), asserting the dial path via the SDK `DialEvent`. - Propagates a V2 initiator's graceful half-close to legacy hosts via `edgeXgressConn.FlowFromFabricToXgressClosed`, which emits an edge FIN when the fabric->app half of the circuit closes. The SDK signals half-close to its router xgress peer with the native xgress EOF flag; without translating that to an edge FIN, a legacy host reading to EOF stalled until teardown. - Records the dialing identity id as the circuit `ClientId` for sessionless V2 dials, since there is no dial session to key on; updates `Test_OidcEvents` to match. - Adds `tests/connect_v2_teardown_test.go` covering client- and host-initiated close propagation on both the V2 and forced-V1 paths. - Polls for the asynchronous conn close in the SDK posture-check tests (`awaitClientConnClosed`): revocation tears the circuit down out of band, so checking `IsClosed` immediately after the first read error was racy. - Temporarily pins sdk-golang/v2 to the openziti/sdk-golang#959 commit, which carries the matching xgress conn-close-on-teardown fix the V2 posture tests depend on; to be repointed at the next sdk-golang pre-release before merge. For openziti/sdk-golang#936.
2026-09-10 00:35:41 +00:00 · 2026-03-25 16:58:08 -04:00
parent 86092a8640
commit 67eba590e2
25 changed files with 807 additions and 397 deletions
@@ -1,120 +0,0 @@
-# Bind Message Channel Ordering Analysis
-
-## Overview
-
-With multi-underlay (separate control and data channels), messages sent on different
-channels have no ordering guarantee relative to each other. This document analyzes
-the channel usage for bind-related messages and identifies race condition potential.
-
-## Channel Usage Map
-
-### Router → SDK
-
-| Message | Channel | Method | Code |
-|---|---|---|---|
-| StateConnected (bind reply) | **Data** | SendAndWaitForWire | listener.go:997 `GetDefaultSender()` |
-| BindSuccess | **Control** | Send | fabric.go:137 `GetControlSender()` |
-| ConnInspectRequest (post-create) | **Control** | TrySend | hosted.go:628 `GetControlSender()` |
-| ConnInspectRequest (validate) | **Control** | SendForReply | fabric.go:156 `GetControlSender()` |
-| StateClosed (bind error) | **Data** | TrySend | hosted.go:482 `GetDefaultSender()` |
-| Dial | **Control** | SendForReply | dialer.go:171 `GetControlSender()` |
-| StateClosed (conn close) | **Data** | SendAndWaitForWire | via `SendState → GetDefaultSender()` |
-| Data | **Data** | Send | fabric.go:423 `GetDefaultSender()` |
-
-### SDK → Router
-
-| Message | Channel | Method | Code |
-|---|---|---|---|
-| Bind | **Control** | SendForReply | hosting_conn.go:476 `GetControlSender()` |
-| Unbind | **Control** | SendAndWaitForWire | hosting_conn.go:503 `GetControlSender()` |
-| ConnInspectResponse | **Control** | Reply | hosting_conn.go:195 `GetControlSender()` |
-| DialSuccess/Failed | reply (matches sender) | | |
-| StateClosed (conn close) | **Data** | SendAndWaitForWire | via `SendState → GetDefaultSender()` |
-| Data | **Data** | | via `GetDefaultSender()` |
-
-## Reply Mechanism
-
-`SendForReply` matches replies by sequence number via a single global waiter map
-shared across all underlays (`channel/multi.go`). Replies are matched regardless
-of which underlay they arrive on. So the bind request (sent on control) and its
-StateConnected reply (sent on data) are correctly matched.
-
-## Race Condition Analysis
-
-### 1. ConnInspectRequest vs StateConnected — Safe (no bug)
-
-The post-create inspect is queued via `go queuePostCreateInspect(terminator)` right
-after StateConnected is sent. It goes through the event loop + `evaluatePostCreateInspects`
-tick interval. In practice it arrives much later. But there's no guarantee — if the
-data channel is congested and control isn't, the inspect could arrive first.
-
-**Impact if inspect arrives first:** Safe. The `edgeHostConn` is registered in the
-mux *before* `listen()` is even called (factory.go:201). So the inspect finds the
-sink, calls `handleInspect`, gets `ConnTypeBind` — correct answer. The SDK's
-`listen()` is still blocked on `SendForReply` waiting for StateConnected, but
-that's independent.
-
-### 2. BindSuccess vs StateConnected — Safe (no bug)
-
-BindSuccess goes on control, StateConnected goes on data. BindSuccess could arrive
-first. But the handler just sets `conn.established.Store(true)` — no dependency on
-StateConnected having been received.
-
-### 3. StateClosed (bind error) vs StateConnected — Safe (same channel)
-
-If bind access is lost during setup (listener.go:1027), the close is sent on data
-(`GetDefaultSender().TrySend`), same channel as StateConnected. FIFO ordering within
-the data channel means StateConnected arrives first.
-
-### 4. Dial vs bind lifecycle — Naturally ordered
-
-Dials can only happen after the terminator is established in the controller, which
-is much later than bind processing. Not a real race.
-
-### 5. Data vs StateConnected — Naturally ordered
-
-Data can only flow after a dial succeeds, which requires controller establishment.
-The architectural sequencing (bind → establish → dial → data) prevents this race
-regardless of channel assignment.
-
-## Current Comment Analysis
-
-listener.go:996 says:
-```
-// this needs to go on the data channel to ensure it gets there before data gets there or a state closed msg
-```
-
-This concern about ordering with data doesn't hold with multi-underlay anyway —
-if StateConnected were on control and data on data, they'd be independent paths.
-But the concern is moot because data can't arrive until a dial completes (which
-requires controller establishment), so there's a natural architectural ordering
-that makes the channel choice irrelevant for data ordering.
-
-The concern about StateClosed ordering IS valid within the data channel — bind-error
-closes are sent on data after StateConnected, so FIFO ordering within the data
-channel keeps them ordered. But bind-error closes happen before any data flows,
-so there's nothing else on data to race with.
-
-## Recommendation: Use Control Channel for Bind Lifecycle
-
-Most bind lifecycle messages already use the control channel:
- Bind request → control
- Unbind → control
- BindSuccess → control
- ConnInspectRequest/Response → control
- Dial → control
-
-Only two use data:
- StateConnected (bind reply) → data
- StateClosed (bind error) → data
-
-Moving these to control would:
- Give strict FIFO ordering for all bind lifecycle messages on one channel
- Eliminate cross-channel ordering dependencies
- Make the ordering correct by design rather than by accident of timing
- Keep dial-related closes on data (where they must be to not race with data)
-
-The current code has no bugs from these races — the handlers are safe (inspect
-before bind works because edgeHostConn is already in the mux), and the timing
-makes races impractical (data before StateConnected is impossible because dial
-hasn't happened). The change would be a correctness/clarity improvement.
@@ -206,8 +206,9 @@ func DecodeCreateCircuitV2Response(m *channel.Message) (*CreateCircuitV2Response

 // CreateCircuitV3Request is sent from a router to the controller to create a circuit
 // without a service session token. The router has already authorized the dial locally
-// via RDM and provides the identity and service IDs directly, along with a pre-assigned
-// circuit ID.
+// via RDM and provides the identity and service IDs directly. CircuitId may be left
+// empty, in which case the controller generates one (matching V1/V2 behavior); if the
+// router supplies a non-empty value it is honored.
 type CreateCircuitV3Request struct {
 	IdentityId           string
 	ServiceId            string
@@ -300,10 +301,10 @@ func DecodeCreateCircuitV3Request(m *channel.Message) (*CreateCircuitV3Request,
 		return nil, errors.New("no service id provided in create circuit v3 request")
 	}

+	// CircuitId is optional: an empty value tells the controller to generate one
+	// (matching V1/V2 behavior). A non-empty value, when supplied, is honored as
+	// the pre-assigned circuit ID.
 	circuitId, _ := m.GetStringHeader(CreateCircuitV3ReqCircuitIdHeader)
-	if circuitId == "" {
-		return nil, errors.New("no circuit id provided in create circuit v3 request")
-	}

 	apiSessionToken, _ := m.GetStringHeader(CreateCircuitReqApiSessionTokenHeader)

@@ -596,6 +596,10 @@ type RouterDataModel struct {
 	Revocations      cmap.ConcurrentMap[string, *edge_ctrl_pb.DataState_Revocation] `json:"revocations"`
 	cachedPublicKeys concurrenz.AtomicValue[map[string]crypto.PublicKey]

+	// serviceNameIndex maps service name -> service id, kept in sync with Services.
+	// Controller enforces unique service names, so this is a 1:1 mapping.
+	serviceNameIndex cmap.ConcurrentMap[string, string]
+
 	terminatorIdCache cmap.ConcurrentMap[string, string]

 	lastSaveIndex *uint64
@@ -649,6 +653,7 @@ func NewBareRouterDataModel(routerId string) *RouterDataModel {
 		terminatorIdCache: cmap.New[string](),
 		subscriptions:     cmap.New[*IdentitySubscription](),
 		selfRouterId:      routerId,
+		serviceNameIndex:  cmap.New[string](),
 	}
 }

@@ -675,6 +680,7 @@ func NewReceiverRouterDataModel(routerId string, closeNotify <-chan struct{}) *R
 		stopNotify:               make(chan struct{}),
 		terminatorIdCache:        cmap.New[string](),
 		selfRouterId:             routerId,
+		serviceNameIndex:         cmap.New[string](),
 	}
 	go result.processSubscriberEvents()
 	return result
@@ -703,6 +709,7 @@ func NewReceiverRouterDataModelFromDataState(routerId string, dataState *edge_ct
 		timelineId:               dataState.TimelineId,
 		terminatorIdCache:        cmap.New[string](),
 		selfRouterId:             routerId,
+		serviceNameIndex:         cmap.New[string](),
 	}

 	if tIdCache, ok := dataState.Caches[edge_ctrl_pb.CacheType_TerminatorIds.String()]; ok && tIdCache != nil && tIdCache.Data != nil {
@@ -747,6 +754,7 @@ func NewReceiverRouterDataModelFromExisting(routerId string, existing *RouterDat
 		timelineId:               existing.timelineId,
 		terminatorIdCache:        existing.terminatorIdCache,
 		selfRouterId:             routerId,
+		serviceNameIndex:         existing.serviceNameIndex,
 	}
 	currentIndex := existing.CurrentIndex()
 	result.SetCurrentIndex(currentIndex)
@@ -1167,8 +1175,10 @@ func (rdm *RouterDataModel) HandleServiceEvent(index uint64, event *edge_ctrl_pb

 	if event.Action == edge_ctrl_pb.DataState_Delete {
 		var cleanupActions []*edge_ctrl_pb.DataState_Event
+		var removedName string
 		rdm.Services.RemoveCb(model.Service.Id, func(key string, v *Service, exists bool) bool {
 			if exists {
+				removedName = v.Name
 				removeFromConfigs(v.Configs)

 				v.servicePolicies.IterCb(func(servicePolicyId string, _ struct{}) {
@@ -1194,6 +1204,14 @@ func (rdm *RouterDataModel) HandleServiceEvent(index uint64, event *edge_ctrl_pb
 			return exists
 		})

+		if removedName != "" {
+			// Only drop the index entry if it still points at us; a new service with the
+			// same name might have been created between our RemoveCb and this step.
+			rdm.serviceNameIndex.RemoveCb(removedName, func(key string, v string, exists bool) bool {
+				return exists && v == model.Service.Id
+			})
+		}
+
 		for _, cleanupAction := range cleanupActions {
 			rdm.Handle(index, cleanupAction)
 		}
@@ -1206,6 +1224,7 @@ func (rdm *RouterDataModel) HandleServiceEvent(index uint64, event *edge_ctrl_pb
 			index:              index,
 		}

+		var oldName string
 		rdm.Services.Upsert(model.Service.Id, updatedService, func(exist bool, valueInMap *Service, newValue *Service) *Service {
 			var configsToRemove []string
 			var configsToAdd []string
@@ -1213,6 +1232,7 @@ func (rdm *RouterDataModel) HandleServiceEvent(index uint64, event *edge_ctrl_pb
 			if !exist {
 				configsToAdd = newValue.Configs
 			} else {
+				oldName = valueInMap.Name
 				configsToRemove, configsToAdd = diffStringSlices(valueInMap.Configs, newValue.Configs)
 			}

@@ -1228,6 +1248,15 @@ func (rdm *RouterDataModel) HandleServiceEvent(index uint64, event *edge_ctrl_pb
 			return newValue
 		})

+		// Keep the name -> id index in sync. On rename, only remove the old mapping
+		// if it still points at this service (same guard as on delete).
+		if oldName != "" && oldName != updatedService.Name {
+			rdm.serviceNameIndex.RemoveCb(oldName, func(key string, v string, exists bool) bool {
+				return exists && v == updatedService.Id
+			})
+		}
+		rdm.serviceNameIndex.Set(updatedService.Name, updatedService.Id)
+
 		updatedService.servicePolicies.IterCb(func(servicePolicyId string, _ struct{}) {
 			rdm.NotifyServicePolicyServiceChange(servicePolicyId, index)
 		})
@@ -2010,6 +2039,14 @@ func (rdm *RouterDataModel) withService(serviceId string, f func(service *Servic
 	}
 }

+// ServiceIdByName returns the ID of the service with the given name, and a
+// boolean indicating whether an entry was found. Uses an index kept in sync
+// with Services at the mutation sites. Controller enforces unique service
+// names, so this is a 1:1 lookup.
+func (rdm *RouterDataModel) ServiceIdByName(name string) (string, bool) {
+	return rdm.serviceNameIndex.Get(name)
+}
+
 func (rdm *RouterDataModel) withServicePolicy(servicePolicyId string, f func(servicePolicy *ServicePolicy)) {
 	if servicePolicy, _ := rdm.ServicePolicies.Get(servicePolicyId); servicePolicy != nil {
 		f(servicePolicy)
@@ -2532,6 +2569,33 @@ func (rdm *RouterDataModel) Validate(correct *RouterDataModel, sink DiffSink) {
 	rdm.subscriptions.IterCb(func(key string, v *IdentitySubscription) {
 		v.Diff(rdm, false, sink)
 	})
+	rdm.validateServiceNameIndex(sink)
+}
+
+// validateServiceNameIndex reports any inconsistency between Services and the
+// private serviceNameIndex. Each service must have a name -> id entry that
+// points back at it, and each index entry must reference an existing service
+// whose Name equals the indexed key.
+func (rdm *RouterDataModel) validateServiceNameIndex(sink DiffSink) {
+	const et = "service-name-index"
+
+	rdm.Services.IterCb(func(id string, svc *Service) {
+		indexed, ok := rdm.serviceNameIndex.Get(svc.Name)
+		if !ok {
+			sink(et, svc.Name, DiffTypeSub, fmt.Sprintf("missing index entry for service %q (id=%s)", svc.Name, id))
+		} else if indexed != id {
+			sink(et, svc.Name, DiffTypeMod, fmt.Sprintf("index entry for %q points at %s, expected %s", svc.Name, indexed, id))
+		}
+	})
+
+	rdm.serviceNameIndex.IterCb(func(name, id string) {
+		svc, ok := rdm.Services.Get(id)
+		if !ok {
+			sink(et, name, DiffTypeAdd, fmt.Sprintf("index entry %q -> %s but no such service", name, id))
+		} else if svc.Name != name {
+			sink(et, name, DiffTypeMod, fmt.Sprintf("index key %q maps to service %s but service name is %q", name, id, svc.Name))
+		}
+	})
 }

 func (rdm *RouterDataModel) Diff(o *RouterDataModel, sink DiffSink) {
@@ -35,8 +35,9 @@ import (

 // NewCreateCircuitV3Handler creates a handler for CreateCircuitV3 requests. These requests
 // come from routers that have already authorized the dial locally via RDM, so no service
-// session token is required. Instead, the request carries identity ID, service ID, and
-// a pre-assigned circuit ID.
+// session token is required. Instead, the request carries the identity ID and service ID,
+// and either a pre-assigned circuit ID or an empty one, in which case the controller
+// generates it.
 func NewCreateCircuitV3Handler(appEnv *env.AppEnv, ch channel.Channel) channel.ContentTypeReceiver {
 	handler := &createCircuitHandler{
 		baseRequestHandler: baseRequestHandler{
@@ -71,7 +71,7 @@ require (
 	github.com/openziti/jwks v1.0.6
 	github.com/openziti/metrics v1.4.5
 	github.com/openziti/runzmd v1.0.90
-	github.com/openziti/sdk-golang/v2 v2.0.0-pre1
+	github.com/openziti/sdk-golang/v2 v2.0.0-pre1.0.20260624192439-7081a4d51c11
 	github.com/openziti/secretstream v0.1.51
 	github.com/openziti/transport/v2 v2.0.216
 	github.com/openziti/x509-claims v1.0.3
@@ -554,6 +554,8 @@ github.com/openziti/runzmd v1.0.90 h1:fasGlaq9xV+zohEGDC7Q0nOLA0n8Kpfccribc+VGQV
 github.com/openziti/runzmd v1.0.90/go.mod h1:ma3b7UdVYAC9ZCVUSjevUH/7yz9asI537XPsEh7+j14=
 github.com/openziti/sdk-golang/v2 v2.0.0-pre1 h1:oB875uZ+oRaIR5s4CmGH116fo1L7klb+LaHlhA7V60k=
 github.com/openziti/sdk-golang/v2 v2.0.0-pre1/go.mod h1:y0Kvj1jQ6FITI64T5mHAx64rtrZaXLCY1NaFrSRUboI=
+github.com/openziti/sdk-golang/v2 v2.0.0-pre1.0.20260624192439-7081a4d51c11 h1:KfLYEDZfsHVuVdlKvmphoWWpbZNea1RZn93JbZenSBo=
+github.com/openziti/sdk-golang/v2 v2.0.0-pre1.0.20260624192439-7081a4d51c11/go.mod h1:y0Kvj1jQ6FITI64T5mHAx64rtrZaXLCY1NaFrSRUboI=
 github.com/openziti/secretstream v0.1.51 h1:j/rMfIzBNqZD5a1EKV8J4Z5QeaoSK56s/zxvvB08eSA=
 github.com/openziti/secretstream v0.1.51/go.mod h1:YapZv2c/SyZyohn6Q0MJkl8SUuBbs9Q6XWSiboBc1jA=
 github.com/openziti/transport/v2 v2.0.216 h1:/2ALqUaeDzOfvZwzGSPNtWqMu8srgA1AERF3Nk94nDQ=
@@ -1,157 +0,0 @@
-# Router Data Model Race Conditions
-
-## Context
-
-Investigation of missing terminators in sdk-hosting-test. 14 out of 15000 expected terminators were
-missing. 6 identities on router-ap-0 were told they lost bind access to some services during a full
-RDM state replacement, causing permanent terminator loss (SDK treats "bind access lost" as
-`RetryNotRetriable`).
-
-Timeline:
- 23:13:10 - CLUSTER_NO_LEADER errors
- 23:13:11.931 - Full RDM state received from ctrl3 (index 205116, replacing old index 202533)
- 23:13:11.990 - RDM replacement starts
- 23:13:11.994 - RDM replacement complete
- 23:13:12.016-018 - Bind access lost events for 6 identities across 5 services
-
-The access loss was TEMPORARY - current RDM at index 210874 shows correct data for all affected
-identities. The router likely re-subscribed to a different controller with a correct sender model.
-
---
-
-## Race Condition 1: SyncAllSubscribers was async (FIXED)
-
-### Description
-
-`SyncAllSubscribers()` only queued a `syncAllSubscribersEvent` to the events channel (async). After
-`SetRouterDataModel` stored the new model and returned, the pool worker became free to process
-incremental `ApplyChangeSet` calls. Meanwhile, the `processSubscriberEvents` goroutine was still
-processing the `syncAllSubscribersEvent` on a separate goroutine.
-
-This meant incremental changes could modify the model's ServiceAccess data concurrently with
-`checkForChanges` iterating it, potentially causing false `ServiceAccessLostEvent` notifications.
-
-### Fix
-
-Made `SyncAllSubscribers()` synchronous by adding a `completeNotify` channel that blocks until
-the sync event is fully processed.
-
-### Relevance to incident
-
-Likely insufficient to explain this specific incident since only terminator updates (not policy
-changes) should have been happening. Terminator updates don't flow through the RDM.
-
---
-
-## Race Condition 2: Controller BuildAll / Entity Constraint Registration Gap
-
-### Description
-
-In `InstantStrategy.Initialize()` (sync_instant.go:139-244), the ordering is:
-
-```
-1. Line 140:      Create sender model
-2. Line 155:      BuildAll (reads DB in View/MVCC snapshot at time T_build)
-3. Line 161:      Start handleRouterModelEvents goroutine + NewListener
-4. Lines 163-241: Register entity constraints (addToChangeSet handlers)
-5. Line 242:      Register tx complete listener (completeChangeSet)
-```
-
-`raft.NewRaft()` is called BEFORE `Initialize()`. The Raft subsystem is already running and can
-apply NEW log entries (index > startIndex) via `FSM.Apply` at any time.
-
-**Window 1: Between BuildAll and entity constraint registration (steps 2-4)**
-
-If a Raft entry commits to the DB after BuildAll's MVCC snapshot time but before entity constraints
-are registered:
- The data is written to the DB by `FSM.Apply`
- BuildAll's MVCC snapshot doesn't see it
- Entity constraints not registered => no `addToChangeSet` fires => no events generated
- Sender model **permanently** misses this data
-
-**Window 2: Between entity constraint and tx complete listener registration (steps 4-5)**
-
-If a Raft entry goes through while entity constraints are registered but `completeChangeSet` isn't:
- Entity constraints fire => `addToChangeSet` accumulates events in `strategy.changeSets[index]`
- Transaction completes but `completeChangeSet` not registered => events sit in the map
- When the NEXT transaction completes (after step 5), `completeChangeSet` runs:
-  ```go
-  for k := range strategy.changeSets {
-      if k <= index {
-          delete(strategy.changeSets, k)  // DELETED without being applied!
-      }
-  }
-  ```
- The orphaned changeSets are silently cleaned up
-
-### Impact
-
-If ctrl3's sender model was built during startup while Raft entries containing identity-to-policy
-associations were being applied, the sender model would permanently miss those associations.
-
-When ctrl3 sends a full state snapshot to a router:
- Identities would be present
- Service policies would be present
- But `ServicePolicyChange` (RelatedIdentity) events linking specific identities to policies would
-  be missing
-
-The router would build its RDM from this incomplete state, `checkForChanges` would find services
-missing from the new ServiceAccess, and fire `ServiceAccessLostEvent`.
-
-### Validation gap
-
-`ValidateServicePolicies` (sync_instant.go:1179-1197) does NOT validate `SenderIdentity.ServicePolicies`
-associations. It only validates the policy's Services and PostureChecks maps. So this inconsistency
-would not be caught by validation.
-
-### Open questions
-
- Was ctrl3 recently (re)started before the incident?
- Were there any Raft entries being applied during ctrl3's `Initialize()` call?
- The CLUSTER_NO_LEADER at 23:13:10 is suspicious - did it cause a controller restart?
-
---
-
-## Race Condition 3: cmap IterCb during getDataStateAlreadyLocked
-
-### Description
-
-`getDataStateAlreadyLocked` (router_data_model_sender.go:411-541) builds the full state by iterating
-cmaps using `IterCb`. While the EventCache lock prevents new events from being stored/applied, cmap's
-`IterCb` uses per-shard RLock (not a global snapshot).
-
-If there were any concurrent writer to the cmaps outside the EventCache lock path, items could be
-skipped during iteration.
-
-### Assessment
-
-The EventCache lock should prevent this since `ApplyChangeSet` -> `EventCache.Store` acquires the
-same lock. Unless there's a path that modifies `SenderIdentity.ServicePolicies` without going through
-the EventCache (e.g., synthetic events, BuildAll running concurrently), this shouldn't happen.
-
-**Likelihood: Low** unless BuildAll and event application overlap.
-
---
-
-## Additional Note: BoltDbFsm.startIndex never set
-
-`BoltDbFsm.startIndex` (the struct field at fsm.go:88) is never assigned. In `Init()`, the local
-variable `startIndex` is loaded and set to `self.index`, but the struct field stays at 0. So
-`GetStartIndex()` -> `GetStartRaftIndex()` always returns 0, and `RaftIndexProvider` initializes
-with index 0.
-
-This means `BuildAll` uses `indexProvider.CurrentIndex()` = 0, and `SetCurrentIndex(0)` on the
-EventCache is a no-op. All subsequent events are accepted (since any raft index > 0). This appears
-to be a bug, though it may not have functional impact since events arrive in order.
-
---
-
-## Architectural Concern: SDK retry behavior
-
-When the router reports a bind access loss (even a temporary one due to RDM replacement), the SDK
-receives `RetryNotRetriable` and permanently closes the listener. The SDK polls the controller for
-service changes independently, so it doesn't know the loss was temporary.
-
-Suggested improvement: defer sending retry hints to the SDK until the SDK is getting service updates
-from the router using the same subscriber mechanism, so temporary blips during RDM replacement don't
-cause permanent terminator loss.
@@ -67,9 +67,15 @@ type RemoveListener func()
 // ConnState encapsulates the authentication and authorization context for an
 // edge connection, bundling API session credentials, service-specific tokens,
 // and policy enforcement metadata for streamlined access control decisions.
+//
+// ServiceId is the authoritative service identifier for the connection. For
+// V1 dials it matches ServiceSessionToken.ServiceId; for V2 sessionless
+// dials there is no token so this field is the only carrier. Set unconditionally
+// by the connect handlers.
 type ConnState struct {
 	ApiSessionToken     *ApiSessionToken
 	ServiceSessionToken *ServiceSessionToken
+	ServiceId           string
 	PolicyType          edge_ctrl_pb.PolicyType
 }

@@ -1224,7 +1230,14 @@ func (self *ManagerImpl) RemoveLegacyServiceSession(serviceSessionToken *Service
 		edgeConn, connIdToSink := GetConnProviderAndSinksFromCh(activeChannel)

 		for connId, sink := range connIdToSink {
-			if serviceSessionToken.TokenId() == sink.GetData().ServiceSessionToken.TokenId() {
+			// V2 (sessionless) dials register a sink whose ConnState has no
+			// ServiceSessionToken; such a conn can't belong to a legacy service
+			// session, so skip it rather than dereferencing a nil token.
+			sinkToken := sink.GetData().ServiceSessionToken
+			if sinkToken == nil {
+				continue
+			}
+			if serviceSessionToken.TokenId() == sinkToken.TokenId() {
 				err := edgeConn.CloseConn(connId, fmt.Sprintf("closing connId %d, legacy service session was removed by controller sync", connId))

 				if err != nil {
@@ -77,6 +77,13 @@ func (self *Acceptor) BindChannel(binding channel.Binding) error {
 		},
 	})

+	channel.AddReceiveHandlers(binding, &channel.AsyncFunctionReceiveAdapter{
+		Type: sdkEdge.ContentTypeConnectV2,
+		Handler: func(m *channel.Message, ch channel.Channel) {
+			conn.processConnectV2(m, ch)
+		},
+	})
+
 	channel.AddReceiveHandlers(binding, &channel.AsyncFunctionReceiveAdapter{
 		Type: sdkEdge.ContentTypeBind,
 		Handler: func(m *channel.Message, ch channel.Channel) {
@@ -136,6 +143,7 @@ func (self *Acceptor) BindChannel(binding channel.Binding) error {
 	binding.AddReceiveHandlerF(sdkEdge.ContentTypeXgPayload, conn.handleXgPayload)
 	binding.AddReceiveHandlerF(sdkEdge.ContentTypeXgAcknowledgement, conn.handleXgAcknowledgement)
 	binding.AddReceiveHandlerF(sdkEdge.ContentTypeXgClose, conn.handleXgClose)
+	binding.AddReceiveHandlerF(sdkEdge.ContentTypeXgControl, conn.handleXgControl)

 	// Since data is the most common type, usually it gets to dispatch directly.
 	// For now, we use handleDataMessage instead of the mux directly so we can log
@@ -261,6 +261,7 @@ const (
 	FlagPostCreateAccessChecked = 1
 	FlagIsCircuitInitiator      = 2
 	FlagIsHostSide              = 3
+	FlagSentFin                 = 4

 	FlagPostCreateAccessCheckedMask = 1 << FlagPostCreateAccessChecked
 	FlagIsCircuitInitiatorMask      = 1 << FlagIsCircuitInitiator
@@ -284,7 +285,16 @@ func (self *edgeXgressConn) GetCircuitId() string {
 }

 func (self *edgeXgressConn) GetServiceId() string {
-	if data := self.GetData(); data != nil && data.ServiceSessionToken != nil {
+	data := self.GetData()
+	if data == nil {
+		return ""
+	}
+	if data.ServiceId != "" {
+		return data.ServiceId
+	}
+	// Defensive fallback for any path that didn't populate ConnState.ServiceId
+	// directly but did set the legacy ServiceSessionToken.
+	if data.ServiceSessionToken != nil {
 		return data.ServiceSessionToken.ServiceId
 	}
 	return ""
@@ -499,6 +509,36 @@ func (self *edgeXgressConn) close(notify bool, reason string) {
 	}
 }

+// FlowFromFabricToXgressClosed implements xgress.SignalConnection. It is invoked
+// when the fabric->app (tx) half of the circuit closes gracefully: the far side
+// sent EOF/end-of-circuit without tearing the whole circuit down. This happens
+// on the ConnectV2 path when an xgress-based initiator closes (the SDK sends a
+// half-close EOF rather than a full CircuitEnd), with this conn bridging to a
+// legacy host. We propagate a FIN to the SDK edge conn so a blocked Read returns
+// io.EOF, while leaving the app->fabric (rx) half open so the conn can still
+// write. This is the half-close counterpart to close(), which sends a full
+// StateClosed and tears down both halves.
+func (self *edgeXgressConn) FlowFromFabricToXgressClosed() {
+	if self.flags.IsSet(FlagClosed) || self.GetChannel().IsClosed() {
+		return
+	}
+
+	// Send the FIN at most once; a subsequent full close still sends StateClosed.
+	if !self.flags.CompareAndSet(FlagSentFin, false, true) {
+		return
+	}
+
+	// An empty data message carrying the FIN flag; the SDK's chunk reader maps
+	// FIN to io.EOF. Sent on the default (data) sender so it stays ordered
+	// behind any payloads already forwarded to the SDK.
+	msg := edge.NewDataMsg(self.Id(), nil)
+	msg.PutUint32Header(edge.FlagsHeader, edge.FIN)
+	if err := self.GetDefaultSender().Send(msg); err != nil {
+		pfxlog.ContextLogger(self.GetChannel().Label()).WithField("connId", self.Id()).
+			WithError(err).Warn("unable to send FIN to edge client on fabric-to-xgress close")
+	}
+}
+
 func (self *edgeXgressConn) AcceptMessage(msg *channel.Message, _ edge.SdkChannel) {
 	if msg.ContentType == edge.ContentTypeTraceRoute {
 		headers := channel.Headers{}
@@ -18,6 +18,7 @@ package xgress_edge

 import (
 	"fmt"
+	"math/big"
 	"strings"
 	"time"

@@ -167,10 +168,14 @@ func (factory *Factory) CreateListener(optionsData xgress.OptionsData) (xgress_r
 		return nil, fmt.Errorf("could not generate version header: %v", err)
 	}

+	capMask := &big.Int{}
+	capMask.SetBit(capMask, edge.RouterCapabilityConnectV2, 1)
+
 	headers := map[int32][]byte{
 		channel.HelloVersionHeader:       versionHeader,
 		edge.SupportsBindSuccessHeader:   {1},
 		edge.SupportsPostureChecksHeader: {1},
+		edge.RouterCapabilitiesHeader:    capMask.Bytes(),
 	}

 	wrappedId := state.WrapIdentityWithCertValidation(factory.env.GetRouterId(), factory.stateManager)
@@ -902,7 +902,7 @@ func (self *edgeClientConn) processConnect(req *channel.Message, ch channel.Chan

 	connectCtx.ServiceSessionToken = serviceSessionToken

-	if err = self.checkAccess(serviceSessionToken.ServiceId, edge_ctrl_pb.PolicyType_DialPolicy); err != nil {
+	if err = self.checkAccessIfOidc(serviceSessionToken.ServiceId, edge_ctrl_pb.PolicyType_DialPolicy); err != nil {
 		log.WithError(err).Error("access denied")
 		self.sendStateClosedReply(err.Error(), req)
 		return
@@ -961,6 +961,163 @@ func (self *edgeClientConn) processConnect(req *channel.Message, ch channel.Chan
 	handler.FinishConnect(connectCtx, response, err)
 }

+func (self *edgeClientConn) processConnectV2(req *channel.Message, ch channel.Channel) {
+	log := pfxlog.ContextLogger(ch.Label()).
+		WithFields(sdkedge.GetLoggerFields(req)).
+		WithField("identityId", self.getIdentityId())
+
+	connId, found := req.GetUint32Header(sdkedge.ConnIdHeader)
+	if !found {
+		pfxlog.Logger().Error("connId not set. unable to process connectv2 message")
+		self.sendStateClosedReply("connId not set, required", req)
+		return
+	}
+
+	// connect-v2 is OIDC-only: it authorizes the dial against router-local RDM
+	// state (policy + posture), which only exists for OIDC sessions. A
+	// legacy/non-OIDC session has its posture evaluated at the controller during
+	// session creation, so it must use the V1 dial path. Reject here so a
+	// non-OIDC session cannot bypass posture checks by dialing via V2.
+	if !self.apiSessionToken.IsOidc() {
+		log.Error("connect-v2 requires an OIDC API session; legacy sessions must use the V1 dial path")
+		self.sendStateClosedReply("connect-v2 requires an OIDC API session", req)
+		return
+	}
+
+	serviceIdOrName, found := req.GetStringHeader(sdkedge.ServiceIdHeader)
+	if !found || serviceIdOrName == "" {
+		log.Error("service header not set in ConnectV2 request")
+		self.sendStateClosedReply("service header not set, required", req)
+		return
+	}
+
+	// Resolve service ID
+	identifierType, _ := req.GetByteHeader(sdkedge.ServiceIdentifierTypeHeader)
+	serviceId := serviceIdOrName
+	if identifierType == byte(sdkedge.ServiceIdentifierByName) {
+		rdm := self.listener.factory.stateManager.RouterDataModel()
+		resolvedId, ok := rdm.ServiceIdByName(serviceIdOrName)
+		if !ok {
+			log.WithField("serviceName", serviceIdOrName).Error("service not found by name")
+			self.sendStateClosedReply("service not found", req)
+			return
+		}
+		serviceId = resolvedId
+	}
+
+	if err := self.checkAccess(serviceId, edge_ctrl_pb.PolicyType_DialPolicy); err != nil {
+		log.WithError(err).Error("access denied")
+		self.sendStateClosedReply(err.Error(), req)
+		return
+	}
+
+	ctrlCh := self.apiSessionToken.SelectCtrlCh(self.listener.factory.ctrls)
+	if ctrlCh == nil {
+		log.Error("no controller available, cannot create circuit")
+		self.sendStateClosedReply("no controller available, cannot create circuit", req)
+		return
+	}
+
+	self.checkForStateListener()
+
+	connectCtx := &connectContext{
+		SdkConn:   self,
+		Log:       log,
+		Req:       req,
+		ConnId:    connId,
+		CtrlId:    ctrlCh.PeerId(),
+		ServiceId: serviceId,
+	}
+
+	var handler connectHandler
+	if useXgToSdk, _ := req.GetBoolHeader(sdkedge.UseXgressToSdkHeader); useXgToSdk {
+		log.Debug("use sdk xgress set, setting up sdk flow-control connection")
+		handler = &xgEdgeForwarder{
+			edgeClientConn: self,
+			serviceId:      serviceId,
+			ctrlId:         ctrlCh.PeerId(),
+			originator:     xgress.Initiator,
+			metrics:        self.listener.factory.env.GetXgressMetrics(),
+		}
+	} else {
+		handler = &nonXgConnectHandler{}
+	}
+
+	if !handler.Init(connectCtx) {
+		self.sendStateClosedReply("connect handler init failed", req)
+		return
+	}
+
+	// Build peer data from request headers
+	peerData := make(map[uint32][]byte)
+	for k, v := range peerHeaderRequestMappings {
+		if pk, found := req.Headers[int32(k)]; found {
+			peerData[v] = pk
+		}
+	}
+
+	if identityId := self.getIdentityId(); identityId != "" {
+		peerData[ctrl_msg.DialerIdentityIdHeader] = []byte(identityId)
+		if ident, found := self.listener.factory.stateManager.RouterDataModel().Identities.Get(identityId); found {
+			peerData[ctrl_msg.DialerIdentityNameHeader] = []byte(ident.Name)
+		}
+	}
+
+	terminatorIdentity, _ := req.GetStringHeader(sdkedge.TerminatorIdentityHeader)
+
+	request := &ctrl_msg.CreateCircuitV3Request{
+		IdentityId:           self.getIdentityId(),
+		ServiceId:            serviceId,
+		Fingerprints:         self.fingerprints.Prints(),
+		TerminatorInstanceId: terminatorIdentity,
+		PeerData:             peerData,
+		ApiSessionToken:      self.apiSessionToken.Token(),
+	}
+
+	response, err := self.sendCreateCircuitV3Msg(request.ToMessage(), ctrlCh)
+
+	handler.FinishConnect(connectCtx, response, err)
+}
+
+// sendCreateCircuitV3Msg sends a CreateCircuitV3 message and decodes the response.
+// The V3 response is converted to a V2 response since they have the same fields and
+// the connectHandler interface uses CreateCircuitV2Response.
+func (self *edgeClientConn) sendCreateCircuitV3Msg(msg *channel.Message, ctrlCh ctrlchan.CtrlChannel) (*ctrl_msg.CreateCircuitV2Response, error) {
+	timeout := self.listener.options.Options.GetCircuitTimeout
+	resp, err := msg.WithTimeout(timeout).SendForReply(ctrlCh.GetHighPrioritySender())
+	if err != nil {
+		return nil, err
+	}
+	if resp.ContentType == int32(edge_ctrl_pb.ContentType_ErrorType) {
+		errMsg := string(resp.Body)
+		if errMsg == "" {
+			errMsg = "error state returned from controller with no message"
+		}
+		var circuitResp *ctrl_msg.CreateCircuitV2Response
+		if circuitId, found := resp.GetStringHeader(sdkedge.CircuitIdHeader); found {
+			circuitResp = &ctrl_msg.CreateCircuitV2Response{CircuitId: circuitId}
+		}
+		return circuitResp, errors.New(errMsg)
+	}
+
+	if resp.ContentType != int32(edge_ctrl_pb.ContentType_CreateCircuitV3ResponseType) {
+		return nil, errors.Errorf("unexpected response type %v to request. expected %v",
+			resp.ContentType, edge_ctrl_pb.ContentType_CreateCircuitV3ResponseType)
+	}
+
+	v3Resp, err := ctrl_msg.DecodeCreateCircuitV3Response(resp)
+	if err != nil {
+		return nil, err
+	}
+
+	return &ctrl_msg.CreateCircuitV2Response{
+		CircuitId: v3Resp.CircuitId,
+		Address:   v3Resp.Address,
+		PeerData:  v3Resp.PeerData,
+		Tags:      v3Resp.Tags,
+	}, nil
+}
+
 func (self *edgeClientConn) mapResponsePeerData(m map[uint32][]byte) {
 	for k, v := range peerHeaderRespMappings {
 		if val, ok := m[k]; ok {
@@ -1033,7 +1190,7 @@ func (self *edgeClientConn) processBind(req *channel.Message, ch channel.Channel
 		return
 	}

-	if err = self.checkAccess(serviceSessionToken.ServiceId, edge_ctrl_pb.PolicyType_BindPolicy); err != nil {
+	if err = self.checkAccessIfOidc(serviceSessionToken.ServiceId, edge_ctrl_pb.PolicyType_BindPolicy); err != nil {
 		log.Error(err.Error())
 		self.sendStateClosedReply(err.Error(), req)
 		return
@@ -1188,7 +1345,7 @@ func (self *edgeClientConn) processBindV2(serviceSessionToken *state.ServiceSess
 		}
 	}

-	if err = self.checkAccess(serviceSessionToken.ServiceId, edge_ctrl_pb.PolicyType_BindPolicy); err != nil {
+	if err = self.checkAccessIfOidc(serviceSessionToken.ServiceId, edge_ctrl_pb.PolicyType_BindPolicy); err != nil {
 		log.WithError(err).Error("bind access lost while terminator setup, closing")
 		edgeErr := &EdgeError{
 			Message:   "bind access lost",
@@ -1202,28 +1359,40 @@ func (self *edgeClientConn) processBindV2(serviceSessionToken *state.ServiceSess
 	}
 }

+// checkAccess runs the RDM-based access check (policy + posture) unconditionally.
+// Used by code paths that don't have an equivalent posture-aware check elsewhere
+// (notably the connect-v2 path, where there's no controller-issued service session
+// whose creation would have done the check).
 func (self *edgeClientConn) checkAccess(serviceId string, policyType edge_ctrl_pb.PolicyType) error {
-	if self.apiSessionToken.IsOidc() {
-		stateManager := self.listener.factory.stateManager
-		// if oidc we check on the router, legacy tokens are checked in the controller during terminator creation
-		grantingPolicy, err := stateManager.HasAccess(self.apiSessionToken.IdentityId, self.apiSessionToken.Id, serviceId, policyType)
+	stateManager := self.listener.factory.stateManager
+	grantingPolicy, err := stateManager.HasAccess(self.apiSessionToken.IdentityId, self.apiSessionToken.Id, serviceId, policyType)

-		if err != nil {
-			return err
-		}
+	if err != nil {
+		return err
+	}

-		if grantingPolicy == nil {
-			policyTypeDescriptor := "dial"
-			if policyType == edge_ctrl_pb.PolicyType_BindPolicy {
-				policyTypeDescriptor = "bind"
-			}
-			return errors.Errorf("no access to service, failed %s access check", policyTypeDescriptor)
+	if grantingPolicy == nil {
+		policyTypeDescriptor := "dial"
+		if policyType == edge_ctrl_pb.PolicyType_BindPolicy {
+			policyTypeDescriptor = "bind"
 		}
+		return errors.Errorf("no access to service, failed %s access check", policyTypeDescriptor)
 	}

 	return nil
 }

+// checkAccessIfOidc is checkAccess for paths where legacy (non-OIDC) tokens have
+// their authorization performed elsewhere — historically by the controller at
+// service-session creation time. For OIDC tokens there's no such intermediate
+// step, so the router checks locally via RDM.
+func (self *edgeClientConn) checkAccessIfOidc(serviceId string, policyType edge_ctrl_pb.PolicyType) error {
+	if !self.apiSessionToken.IsOidc() {
+		return nil
+	}
+	return self.checkAccess(serviceId, policyType)
+}
+
 func (self *edgeClientConn) processUnbind(req *channel.Message, ch channel.Channel) {
 	connId, _ := req.GetUint32Header(sdkedge.ConnIdHeader)
 	sessionTokenStr := string(req.Body)
@@ -1606,6 +1775,38 @@ func (self *edgeClientConn) handleXgPayload(msg *channel.Message, _ channel.Chan
 	}
 }

+// handleXgControl handles an xgress control message sent by an SDK-side xgress
+// conn. For trace route requests, the SDK-side channel sequence is stashed into
+// ControlUserVal so the eventual response can be correlated back via
+// ReplyForHeader at this (initiator) router's xgEdgeForwarder.SendControl.
+// For trace route responses, the upstream ControlUserVal must be preserved
+// unchanged — it holds the initiator's request sequence and is what lets the
+// response reach the initiator SDK's SendForReply waiter. Lookup is
+// circuit-id-keyed via xgCircuits.
+func (self *edgeClientConn) handleXgControl(msg *channel.Message, _ channel.Channel) {
+	ctrl, err := xgress.UnmarshallControl(msg)
+	if err != nil {
+		pfxlog.Logger().WithError(err).Error("failed to unmarshal xgress control from sdk")
+		return
+	}
+	if ctrl.Headers == nil {
+		ctrl.Headers = channel.Headers{}
+	}
+	if ctrl.Type == xgress.ControlTypeTraceRoute {
+		ctrl.Headers.PutUint32Header(xgress.ControlUserVal, uint32(msg.Sequence()))
+	}
+
+	edgeFwd, _ := self.xgCircuits.Get(ctrl.CircuitId)
+	if edgeFwd == nil {
+		pfxlog.Logger().WithField("circuitId", ctrl.CircuitId).Error("no edge forwarder found for xgress control")
+		return
+	}
+
+	if err = self.forwarder.ForwardControl(edgeFwd.address, ctrl); err != nil {
+		pfxlog.Logger().WithField("circuitId", ctrl.CircuitId).WithError(err).Error("failed to forward xgress control")
+	}
+}
+
 func (self *edgeClientConn) handleXgAcknowledgement(req *channel.Message, _ channel.Channel) {
 	ack, err := xgress.UnmarshallAcknowledgement(req)
 	if err != nil {
@@ -1644,6 +1845,7 @@ type connectContext struct {
 	CtrlId              string
 	PolicyType          edge_ctrl_pb.PolicyType
 	ServiceSessionToken *state.ServiceSessionToken
+	ServiceId           string // used by ConnectV2 when no ServiceSessionToken is available
 }

 type nonXgConnectHandler struct {
@@ -1657,18 +1859,28 @@ func (self *nonXgConnectHandler) Init(ctx *connectContext) bool {
 		seq:        NewMsgQueue(4),
 	}

+	// V1 carries the service id on the token; V2 carries it directly on ctx.
+	// Either way, ConnState.ServiceId is the single source of truth downstream.
+	serviceId := ctx.ServiceId
+	if ctx.ServiceSessionToken != nil {
+		serviceId = ctx.ServiceSessionToken.ServiceId
+	}
+
 	self.conn.SetData(&state.ConnState{
 		ServiceSessionToken: ctx.ServiceSessionToken,
 		ApiSessionToken:     ctx.SdkConn.apiSessionToken,
+		ServiceId:           serviceId,
 		PolicyType:          edge_ctrl_pb.PolicyType_DialPolicy,
 	})

 	// need to remove session remove listener on close
-	stateManager := ctx.SdkConn.listener.factory.stateManager
-
-	self.conn.onClose = stateManager.AddLegacyServiceSessionRemovedListener(ctx.ServiceSessionToken, func(_ *state.ServiceSessionToken) {
-		self.conn.close(true, "session closed")
-	})
+	// V2 (sessionless) dials have no ServiceSessionToken, so skip the legacy listener
+	if ctx.ServiceSessionToken != nil {
+		stateManager := ctx.SdkConn.listener.factory.stateManager
+		self.conn.onClose = stateManager.AddLegacyServiceSessionRemovedListener(ctx.ServiceSessionToken, func(_ *state.ServiceSessionToken) {
+			self.conn.close(true, "session closed")
+		})
+	}

 	// We can't fix conn id, since it's provided by the client
 	if err := ctx.SdkConn.msgMux.Add(self.conn); err != nil {
@@ -1771,6 +1983,13 @@ func (self *xgEdgeForwarder) SetPostCreateAccessCheckDone() {
 func (self *xgEdgeForwarder) SendPayload(payload *xgress.Payload, timeout time.Duration, _ xgress.PayloadType) error {
 	msg := payload.Marshall()
 	msg.PutUint32Header(sdkedge.ConnIdHeader, self.connId)
+
+	// Track liveness for the forwarder's unroute scheduler regardless of which
+	// send path we take. The fast-path (timeout == 0, used for normal
+	// forwarding via Forwarder.ForwardPayload) previously returned without
+	// updating lastRx, making active circuits look idle to the unroute timer.
+	self.lastRx.Store(time.Now().UnixMilli())
+
 	if timeout == 0 {
 		sent, err := self.ch.GetDefaultSender().TrySend(msg)
 		if err == nil && !sent {
@@ -1785,8 +2004,6 @@ func (self *xgEdgeForwarder) SendPayload(payload *xgress.Payload, timeout time.D
 		return err
 	}

-	self.lastRx.Store(time.Now().UnixMilli())
-
 	if err := msg.WithTimeout(timeout).Send(self.ch.GetDefaultSender()); err != nil {
 		self.listener.droppedMsgMeter.Mark(1)
 		self.listener.droppedPayloadsMeter.Mark(1)
@@ -1817,6 +2034,15 @@ func (self *xgEdgeForwarder) SendAcknowledgement(ack *xgress.Acknowledgement) er
 func (self *xgEdgeForwarder) SendControl(ctrl *xgress.Control) error {
 	msg := ctrl.Marshall()
 	msg.PutUint32Header(sdkedge.ConnIdHeader, self.connId)
+	// For trace route responses, the SDK is waiting via SendForReply on the
+	// original request's channel sequence. The request's sequence was stashed
+	// into ControlUserVal by handleXgControl; promote it back to ReplyForHeader
+	// here so the channel layer matches the reply to the waiter.
+	if ctrl.Type == xgress.ControlTypeTraceRouteResponse {
+		if userVal, ok := ctrl.Headers.GetUint32Header(xgress.ControlUserVal); ok {
+			msg.PutUint32Header(channel.ReplyForHeader, userVal)
+		}
+	}
 	sent, err := self.ch.GetDefaultSender().TrySend(msg)
 	if err == nil && !sent {
 		self.listener.droppedMsgMeter.Mark(1)
@@ -1880,8 +2106,14 @@ func (self *xgEdgeForwarder) FinishConnect(ctx *connectContext, response *ctrl_m
 		msg.Headers[int32(k)] = v
 	}

-	// this needs to go on the data channel to ensure it gets there before data gets there or a state closed msg
-	if err = msg.WithTimeout(5 * time.Second).SendAndWaitForWire(self.ch.GetControlSender()); err != nil {
+	// Must go on the default (data) sender, not the control sender. With
+	// multi-underlay channels the two are independently ordered; if
+	// state_connected went via the control sender, an early terminator-side
+	// payload arriving on the data sender could reach the SDK first, before
+	// it has registered its mux sink, and be dropped. Sending it on the data
+	// sender keeps it ordered ahead of the payloads that follow on the same
+	// sender.
+	if err = msg.WithTimeout(5 * time.Second).SendAndWaitForWire(self.ch.GetDefaultSender()); err != nil {
 		pfxlog.Logger().WithFields(sdkedge.GetLoggerFields(msg)).WithError(err).Error("failed to send state response")
 	}
 }
@@ -1892,6 +2124,12 @@ func (self *xgEdgeForwarder) Unrouted() {
 	defer pfxlog.Logger().WithField("circuitId", self.circuitId).Debug("unroute: complete")
 	self.xgCircuits.Remove(self.circuitId)

+	// Send state_closed at standard priority on the default (data) sender so it
+	// stays ordered behind any payloads already queued for the SDK rather than
+	// leapfrogging them. The destination is unregistered before Unrouted runs,
+	// so no further payloads will be forwarded; FIFO ordering then guarantees the
+	// SDK drains all in-band data before it sees the close. (Unrouted is the
+	// ungraceful teardown path, so we don't synthesize an in-band end-of-circuit.)
 	msg := sdkedge.NewStateClosedMsg(self.connId, "xgress unrouted")
 	err := msg.WithTimeout(5 * time.Second).SendAndWaitForWire(self.ch.GetDefaultSender())
 	if err != nil {
@@ -20,8 +20,8 @@ func (n noopMetrics) PayloadWritten(time.Duration)   {}
 func (n noopMetrics) BufferUnblocked(time.Duration)  {}
 func (n noopMetrics) SendPayloadBuffered(int64)      {}
 func (n noopMetrics) SendPayloadDelivered(int64)     {}
-func (n noopMetrics) MarkRetransmission()             {}
-func (n noopMetrics) MarkRetransmissionFailure()      {}
+func (n noopMetrics) MarkRetransmission()            {}
+func (n noopMetrics) MarkRetransmissionFailure()     {}

 type MockEnv struct {
 	payloadIngester *xgress.PayloadIngester
@@ -0,0 +1,178 @@
+//go:build dataflow
+
+/*
+	Copyright NetFoundry Inc.
+
+	Licensed 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
+
+	https://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.
+*/
+
+package tests
+
+import (
+	"errors"
+	"fmt"
+	"io"
+	"testing"
+	"time"
+
+	"github.com/openziti/sdk-golang/v2/ziti"
+	"github.com/openziti/sdk-golang/v2/ziti/edge"
+	"github.com/openziti/ziti/v2/common/eid"
+	"github.com/openziti/ziti/v2/controller/xt_smartrouting"
+)
+
+// Test_ConnectV2_TeardownPropagation verifies that connection close propagates
+// across the circuit in both directions on the ConnectV2 dial path, and that
+// the V1 fallback still behaves identically.
+//
+// The interesting case is client-initiated close: when the dialing (initiator)
+// side closes, the hosting (terminator) side's blocked Read must return io.EOF
+// promptly, not hang until the channel is torn down. ConnectV2 dials the
+// initiator over SDK xgress while the host stays on the legacy edge conn, so
+// the initiator's end-of-circuit has to cross that boundary and surface as EOF
+// on the host. The matching server-initiated direction already works via the
+// data-flow tests; it's included here as a guard.
+func Test_ConnectV2_TeardownPropagation(t *testing.T) {
+	t.Run("connect-v2", func(t *testing.T) {
+		testTeardownPropagation(t, false)
+	})
+	t.Run("connect-v1-fallback", func(t *testing.T) {
+		testTeardownPropagation(t, true)
+	})
+}
+
+func testTeardownPropagation(t *testing.T, forceV1 bool) {
+	ctx := NewTestContext(t)
+	defer ctx.Teardown()
+	ctx.StartServer()
+	ctx.RequireAdminManagementApiLogin()
+
+	service := ctx.AdminManagementSession.RequireNewServiceAccessibleToAll(xt_smartrouting.Name)
+
+	ctx.CreateEnrollAndStartEdgeRouter()
+
+	_, hostContext := ctx.AdminManagementSession.RequireCreateSdkContext()
+	defer hostContext.Close()
+
+	listener, err := hostContext.Listen(service.Name)
+	ctx.Req.NoError(err)
+	defer listener.Close()
+
+	_, clientContext := ctx.AdminManagementSession.RequireCreateSdkContext()
+	defer clientContext.Close()
+
+	var dialEvt ziti.DialEvent
+	dialEvtSet := false
+	removeListener := clientContext.Events().AddDialListener(func(_ ziti.Context, evt ziti.DialEvent) {
+		if evt.ServiceName == service.Name {
+			dialEvt = evt
+			dialEvtSet = true
+		}
+	})
+	defer removeListener()
+
+	dialOptions := &ziti.DialOptions{ConnectTimeout: 5 * time.Second}
+	if forceV1 {
+		dialOptions.ForceConnectV1 = &forceV1
+	}
+
+	expectedProtocol := edge.DialProtocolConnectV2
+	if forceV1 {
+		expectedProtocol = edge.DialProtocolConnectV1
+	}
+
+	t.Run("client-initiated close surfaces EOF on the host", func(t *testing.T) {
+		errC := make(chan error, 1)
+
+		go func() {
+			defer func() {
+				if val := recover(); val != nil {
+					if err, ok := val.(error); ok {
+						errC <- err
+					} else {
+						errC <- errors.New(fmt.Sprintf("%v", val))
+					}
+				}
+				close(errC)
+			}()
+
+			conn := ctx.WrapConn(clientContext.DialWithOptions(service.Name, dialOptions))
+			ctx.Req.True(dialEvtSet, "expected a dial event for service %s", service.Name)
+			ctx.Req.Equal(expectedProtocol, dialEvt.Protocol, "unexpected dial protocol")
+
+			name := conn.ReadString(512, time.Second)
+			conn.WriteString("hello, "+name, time.Second)
+			conn.RequireClose()
+		}()
+
+		hostConn := ctx.WrapNetConn(listener.AcceptEdge())
+		name := eid.New()
+		hostConn.WriteString(name, time.Second)
+		hostConn.ReadExpected("hello, "+name, time.Second)
+
+		select {
+		case err := <-errC:
+			ctx.Req.NoError(err)
+		case <-time.After(2 * time.Second):
+			t.Fatal("timed out waiting for client goroutine to finish")
+		}
+
+		// The client has closed. The host's Read must observe EOF promptly,
+		// driven by the propagated end-of-circuit, well before any channel
+		// teardown.
+		ctx.Req.NoError(hostConn.SetReadDeadline(time.Now().Add(2 * time.Second)))
+		n, err := hostConn.Read(make([]byte, 1024))
+		ctx.Req.Equal(0, n)
+		ctx.Req.Equal(io.EOF, err, "host should observe EOF after client close, got %v", err)
+	})
+
+	t.Run("host-initiated close surfaces EOF on the client", func(t *testing.T) {
+		errC := make(chan error, 1)
+
+		var clientConn *TestConn
+		go func() {
+			defer func() {
+				if val := recover(); val != nil {
+					if err, ok := val.(error); ok {
+						errC <- err
+					} else {
+						errC <- errors.New(fmt.Sprintf("%v", val))
+					}
+				}
+				close(errC)
+			}()
+
+			hostConn := ctx.WrapNetConn(listener.AcceptEdge())
+			name := hostConn.ReadString(512, time.Second)
+			hostConn.WriteString("hello, "+name, time.Second)
+			hostConn.RequireClose()
+		}()
+
+		clientConn = ctx.WrapConn(clientContext.DialWithOptions(service.Name, dialOptions))
+		name := eid.New()
+		clientConn.WriteString(name, time.Second)
+		clientConn.ReadExpected("hello, "+name, time.Second)
+
+		select {
+		case err := <-errC:
+			ctx.Req.NoError(err)
+		case <-time.After(2 * time.Second):
+			t.Fatal("timed out waiting for host goroutine to finish")
+		}
+
+		ctx.Req.NoError(clientConn.SetReadDeadline(time.Now().Add(2 * time.Second)))
+		n, err := clientConn.Read(make([]byte, 1024))
+		ctx.Req.Equal(0, n)
+		ctx.Req.Equal(io.EOF, err, "client should observe EOF after host close, got %v", err)
+	})
+}
@@ -0,0 +1,170 @@
+//go:build dataflow
+
+/*
+	Copyright NetFoundry Inc.
+
+	Licensed 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
+
+	https://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.
+*/
+
+package tests
+
+import (
+	"math"
+	"testing"
+	"time"
+
+	"github.com/openziti/sdk-golang/v2/ziti"
+	"github.com/openziti/sdk-golang/v2/ziti/edge"
+	"github.com/openziti/ziti/v2/common/eid"
+	"github.com/openziti/ziti/v2/controller/xt_smartrouting"
+)
+
+// Test_ConnectV2_Dataflow exercises the sessionless ConnectV2 dial path
+// end-to-end. The SDK defaults to V2 whenever the router advertises the
+// capability and `ForceConnectV1` is not set. The dial protocol is asserted
+// explicitly via the DialEvent so a capability/auth negotiation regression
+// fails directly rather than only as a hang or data failure.
+func Test_ConnectV2_Dataflow(t *testing.T) {
+	ctx := NewTestContext(t)
+	defer ctx.Teardown()
+	ctx.StartServer()
+	ctx.RequireAdminManagementApiLogin()
+
+	service := ctx.AdminManagementSession.RequireNewServiceAccessibleToAll(xt_smartrouting.Name)
+
+	ctx.CreateEnrollAndStartEdgeRouter()
+	_, hostContext := ctx.AdminManagementSession.RequireCreateSdkContext()
+	defer hostContext.Close()
+
+	listener, err := hostContext.Listen(service.Name)
+	ctx.Req.NoError(err)
+	defer listener.Close()
+
+	testServer := newTestServer(listener, func(conn *testServerConn) error {
+		for {
+			name, eof := conn.ReadString(math.MaxUint16*4, time.Minute)
+			if eof {
+				return conn.server.close()
+			}
+			if name == "quit" {
+				conn.WriteString("ok", time.Second)
+				return conn.server.close()
+			}
+			conn.WriteString("hello, "+name, time.Second)
+		}
+	})
+	testServer.start()
+
+	_, clientContext := ctx.AdminManagementSession.RequireCreateSdkContext()
+	defer clientContext.Close()
+
+	var dialEvt ziti.DialEvent
+	dialEvtSet := false
+	removeListener := clientContext.Events().AddDialListener(func(_ ziti.Context, evt ziti.DialEvent) {
+		if evt.ServiceName == service.Name {
+			dialEvt = evt
+			dialEvtSet = true
+		}
+	})
+	defer removeListener()
+
+	dialOptions := &ziti.DialOptions{
+		ConnectTimeout: 5 * time.Second,
+	}
+
+	conn := ctx.WrapConn(clientContext.DialWithOptions(service.Name, dialOptions))
+	defer conn.Close()
+
+	ctx.Req.True(dialEvtSet, "expected a dial event for service %s", service.Name)
+	ctx.Req.Equal(edge.DialProtocolConnectV2, dialEvt.Protocol, "expected the dial to take the ConnectV2 path")
+
+	name := eid.New()
+	conn.WriteString(name, time.Second)
+	conn.ReadExpected("hello, "+name, time.Second)
+
+	conn.WriteString("quit", time.Second)
+	conn.ReadExpected("ok", time.Second)
+
+	testServer.waitForDone(ctx, 5*time.Second)
+}
+
+// Test_ConnectV1_Fallback_Dataflow confirms that the V1 fallback path still
+// works after the connect-v2 changes — important because the SDK still uses
+// V1 against routers that don't advertise V2, and the ForceConnectV1 escape
+// hatch is a documented supported option.
+func Test_ConnectV1_Fallback_Dataflow(t *testing.T) {
+	ctx := NewTestContext(t)
+	defer ctx.Teardown()
+	ctx.StartServer()
+	ctx.RequireAdminManagementApiLogin()
+
+	service := ctx.AdminManagementSession.RequireNewServiceAccessibleToAll(xt_smartrouting.Name)
+
+	ctx.CreateEnrollAndStartEdgeRouter()
+	_, hostContext := ctx.AdminManagementSession.RequireCreateSdkContext()
+	defer hostContext.Close()
+
+	listener, err := hostContext.Listen(service.Name)
+	ctx.Req.NoError(err)
+	defer listener.Close()
+
+	testServer := newTestServer(listener, func(conn *testServerConn) error {
+		for {
+			name, eof := conn.ReadString(math.MaxUint16*4, time.Minute)
+			if eof {
+				return conn.server.close()
+			}
+			if name == "quit" {
+				conn.WriteString("ok", time.Second)
+				return conn.server.close()
+			}
+			conn.WriteString("hello, "+name, time.Second)
+		}
+	})
+	testServer.start()
+
+	_, clientContext := ctx.AdminManagementSession.RequireCreateSdkContext()
+	defer clientContext.Close()
+
+	var dialEvt ziti.DialEvent
+	dialEvtSet := false
+	removeListener := clientContext.Events().AddDialListener(func(_ ziti.Context, evt ziti.DialEvent) {
+		if evt.ServiceName == service.Name {
+			dialEvt = evt
+			dialEvtSet = true
+		}
+	})
+	defer removeListener()
+
+	forceV1 := true
+	dialOptions := &ziti.DialOptions{
+		ConnectTimeout: 5 * time.Second,
+		ForceConnectV1: &forceV1,
+	}
+
+	conn := ctx.WrapConn(clientContext.DialWithOptions(service.Name, dialOptions))
+	defer conn.Close()
+
+	ctx.Req.True(dialEvtSet, "expected a dial event for service %s", service.Name)
+	ctx.Req.Equal(edge.DialProtocolConnectV1, dialEvt.Protocol, "expected the dial to take the ConnectV1 fallback path")
+	ctx.Req.True(dialEvt.Forced, "expected the V1 dial to be flagged as forced via ForceConnectV1")
+
+	name := eid.New()
+	conn.WriteString(name, time.Second)
+	conn.ReadExpected("hello, "+name, time.Second)
+
+	conn.WriteString("quit", time.Second)
+	conn.ReadExpected("ok", time.Second)
+
+	testServer.waitForDone(ctx, 5*time.Second)
+}
@@ -164,6 +164,24 @@ func Test_CreateCircuitV3(t *testing.T) {
 		ctx.Req.Error(err)
 	})

+	t.Run("controller-generated circuit ID", func(t *testing.T) {
+		ctx.NextTest(t)
+
+		// An empty CircuitId on the request tells the controller to generate one.
+		// This is the connect-v2 router path's default behavior.
+		req := &ctrl_msg.CreateCircuitV3Request{
+			IdentityId: dialerIdentity.Id,
+			ServiceId:  svc.Id,
+			PeerData:   map[uint32][]byte{},
+		}
+
+		resp, err := sendV3Request(req)
+		ctx.Req.NoError(err)
+		ctx.Req.NotNil(resp)
+		ctx.Req.NotEmpty(resp.CircuitId, "controller should have generated a circuit id")
+		ctx.Req.NotEmpty(resp.Address)
+	})
+
 	t.Run("duplicate circuit ID", func(t *testing.T) {
 		ctx.NextTest(t)

@@ -294,7 +294,9 @@ func Test_OidcEvents(t *testing.T) {
 	ctx.Req.Equal("circuit", circuitEvent.Namespace)
 	ctx.Req.Equal("created", string(circuitEvent.EventType))
 	ctx.Req.Equal(service.Id, circuitEvent.ServiceId)
-	ctx.Req.Equal(edgeSession.Id, circuitEvent.ClientId)
+	// ConnectV2 dials are sessionless, so the circuit's ClientId is the dialing
+	// identity id rather than an edge (dial) session id.
+	ctx.Req.Equal(clientIdentity.Id, circuitEvent.ClientId)

 	timeout := time.Second * 20
 	for i := 0; i < 3; i++ {
@@ -310,7 +312,7 @@ func Test_OidcEvents(t *testing.T) {
 		} else if circuitEvent, ok := evt.(*event.CircuitEvent); ok {
 			ctx.Req.Equal("circuit", circuitEvent.Namespace)
 			ctx.Req.Equal("deleted", string(circuitEvent.EventType))
-			ctx.Req.Equal(edgeSession.Id, circuitEvent.ClientId)
+			ctx.Req.Equal(clientIdentity.Id, circuitEvent.ClientId)
 		} else {
 			ctx.Req.Fail("unexpected event type: %v", reflect.TypeOf(evt))
 		}
@@ -127,22 +127,7 @@ func Test_PostureCheck_SDK_Domain_OIDC(t *testing.T) {
 				currentPostureDomain = "invalid"
 				postureCache.Evaluate()

-				lastReadCount := 0
-				var lastReadErr error
-				count := 0
-				for !clientConn.IsClosed() && count <= 20 {
-					var buff []byte
-
-					//read till end of client buffer
-					lastReadCount, lastReadErr = clientConn.Read(buff)
-
-					if lastReadErr != nil {
-						break
-					}
-
-					time.Sleep(100 * time.Millisecond)
-					count = count + 1
-				}
+				lastReadCount, lastReadErr := awaitClientConnClosed(clientConn)

 				t.Run("closes the connection", func(t *testing.T) {
 					ctx.testContextChanged(t)
@@ -135,22 +135,7 @@ func Test_PostureCheck_SDK_MAC_OIDC(t *testing.T) {
 				currentReportedMacs = invalidMacAddr
 				postureCache.Evaluate()

-				lastReadCount := 0
-				var lastReadErr error
-				count := 0
-				for !clientConn.IsClosed() && count <= 20 {
-					var buff []byte
-
-					//read till end of client buffer
-					lastReadCount, lastReadErr = clientConn.Read(buff)
-
-					if lastReadErr != nil {
-						break
-					}
-
-					time.Sleep(100 * time.Millisecond)
-					count = count + 1
-				}
+				lastReadCount, lastReadErr := awaitClientConnClosed(clientConn)

 				t.Run("closes the connection", func(t *testing.T) {
 					ctx.testContextChanged(t)
@@ -146,22 +146,7 @@ func Test_PostureCheck_SDK_OS_OIDC(t *testing.T) {
 				currentReportingOsInfo = invalidOsInfo
 				postureCache.Evaluate()

-				lastReadCount := 0
-				var lastReadErr error
-				count := 0
-				for !clientConn.IsClosed() && count <= 20 {
-					var buff []byte
-
-					//read till end of client buffer
-					lastReadCount, lastReadErr = clientConn.Read(buff)
-
-					if lastReadErr != nil {
-						break
-					}
-
-					time.Sleep(100 * time.Millisecond)
-					count = count + 1
-				}
+				lastReadCount, lastReadErr := awaitClientConnClosed(clientConn)

 				t.Run("closes the connection", func(t *testing.T) {
 					ctx.testContextChanged(t)
@@ -170,22 +170,7 @@ func Test_PostureCheck_SDK_Process_Multi_OIDC(t *testing.T) {
 				currentProcessInfo = invalidProcessInfo
 				postureCache.Evaluate()

-				lastReadCount := 0
-				var lastReadErr error
-				count := 0
-				for !clientConn.IsClosed() && count <= 20 {
-					var buff []byte
-
-					//read till end of client buffer
-					lastReadCount, lastReadErr = clientConn.Read(buff)
-
-					if lastReadErr != nil {
-						break
-					}
-
-					time.Sleep(100 * time.Millisecond)
-					count = count + 1
-				}
+				lastReadCount, lastReadErr := awaitClientConnClosed(clientConn)

 				t.Run("closes the connection", func(t *testing.T) {
 					ctx.testContextChanged(t)
@@ -168,22 +168,7 @@ func Test_PostureCheck_SDK_Process_OIDC(t *testing.T) {
 				currentProcessInfo = invalidProcessInfo
 				postureCache.Evaluate()

-				lastReadCount := 0
-				var lastReadErr error
-				count := 0
-				for !clientConn.IsClosed() && count <= 20 {
-					var buff []byte
-
-					//read till end of client buffer
-					lastReadCount, lastReadErr = clientConn.Read(buff)
-
-					if lastReadErr != nil {
-						break
-					}
-
-					time.Sleep(100 * time.Millisecond)
-					count = count + 1
-				}
+				lastReadCount, lastReadErr := awaitClientConnClosed(clientConn)

 				t.Run("closes the connection", func(t *testing.T) {
 					ctx.testContextChanged(t)
@@ -67,6 +67,26 @@ func requireConnClosed(ctx *TestContext, conn *TestConn) {
 	ctx.Req.True(conn.IsClosed(), "expected connection to be closed by posture revalidation")
 }

+// awaitClientConnClosed polls a conn until it reports closed, returning the last
+// read result so callers can also assert on it. Posture revocation is
+// asynchronous (the router revokes access, the circuit is torn down, and the
+// close propagates back), so checking IsClosed right after the first read error
+// is too eager; a read is what surfaces the close on these conns.
+func awaitClientConnClosed(conn *TestConn) (int, error) {
+	var n int
+	var err error
+	deadline := time.Now().Add(5 * time.Second)
+	for time.Now().Before(deadline) {
+		var buff []byte
+		n, err = conn.Read(buff)
+		if err != nil && conn.IsClosed() {
+			break
+		}
+		time.Sleep(50 * time.Millisecond)
+	}
+	return n, err
+}
+
 // requireConnUsable asserts a connection survives a posture change: revocation,
 // if it were going to happen, fires within ~100ms of the posture update, so this
 // waits well past that and then confirms the connection is still open and still
@@ -23,7 +23,7 @@ require (
 	github.com/openziti/foundation/v2 v2.0.95
 	github.com/openziti/identity v1.0.133
 	github.com/openziti/metrics v1.4.5
-	github.com/openziti/sdk-golang/v2 v2.0.0-pre1
+	github.com/openziti/sdk-golang/v2 v2.0.0-pre1.0.20260624192439-7081a4d51c11
 	github.com/openziti/transport/v2 v2.0.216
 	github.com/openziti/ziti/v2 v2.0.0
 	github.com/orcaman/concurrent-map/v2 v2.0.1
@@ -631,6 +631,8 @@ github.com/openziti/runzmd v1.0.90 h1:fasGlaq9xV+zohEGDC7Q0nOLA0n8Kpfccribc+VGQV
 github.com/openziti/runzmd v1.0.90/go.mod h1:ma3b7UdVYAC9ZCVUSjevUH/7yz9asI537XPsEh7+j14=
 github.com/openziti/sdk-golang/v2 v2.0.0-pre1 h1:oB875uZ+oRaIR5s4CmGH116fo1L7klb+LaHlhA7V60k=
 github.com/openziti/sdk-golang/v2 v2.0.0-pre1/go.mod h1:y0Kvj1jQ6FITI64T5mHAx64rtrZaXLCY1NaFrSRUboI=
+github.com/openziti/sdk-golang/v2 v2.0.0-pre1.0.20260624192439-7081a4d51c11 h1:KfLYEDZfsHVuVdlKvmphoWWpbZNea1RZn93JbZenSBo=
+github.com/openziti/sdk-golang/v2 v2.0.0-pre1.0.20260624192439-7081a4d51c11/go.mod h1:y0Kvj1jQ6FITI64T5mHAx64rtrZaXLCY1NaFrSRUboI=
 github.com/openziti/secretstream v0.1.51 h1:j/rMfIzBNqZD5a1EKV8J4Z5QeaoSK56s/zxvvB08eSA=
 github.com/openziti/secretstream v0.1.51/go.mod h1:YapZv2c/SyZyohn6Q0MJkl8SUuBbs9Q6XWSiboBc1jA=
 github.com/openziti/transport/v2 v2.0.216 h1:/2ALqUaeDzOfvZwzGSPNtWqMu8srgA1AERF3Nk94nDQ=