Publishing and Subscription Hardening - #1823
Merged
Merged
Conversation
Introduce a test seam for deterministically driving the client's PublishingManager and OpcUaSubscription. ScriptableSubscriptionServiceSet (built on a new DelegatingSubscriptionServiceSet) lets a test script the exact Publish/Republish responses the client observes — sequence numbers, keep-alives, gaps, rollover — while parking pipelined Publish requests and capturing the acknowledgements the client sends back. This provides the foundation for the deterministic bug-reproduction tests targeting publishing-manager reliability. PublishScriptHarnessTest is a smoke test proving the seam delivers a scripted keep-alive and captures the resulting acknowledgement. Also add a TestClient.create overload that exposes the transport config builder, so tests can inject a controllable ExecutorService.
PublishingManager.getTimeoutHint() derives the Publish request's timeoutHint
from revisedPublishingInterval * revisedMaxKeepAliveCount * maxPendingPublishes
* 1.5. timeoutHint is a UInt32 (OPC UA Part 4, 7.32), and the overflow guard at
PublishingManager.java:395 assigned the wrong variable:
if (Double.isInfinite(timeoutHint) || timeoutHint > UInteger.MAX_VALUE) {
maxKeepAlive = 0d; // never read again
}
return uint((long) timeoutHint); // still the oversized value
The oversized value therefore still reached uint(), which range-checks and
throws an unchecked NumberFormatException:
java.lang.NumberFormatException: Value is out of range : 10800000000
at UInteger.rangeCheck(UInteger.java:210)
at UInteger.valueOf(UInteger.java:170)
at Unsigned.uint(Unsigned.java:140)
at PublishingManager.getTimeoutHint(PublishingManager.java:405)
at PublishingManager.sendPublishRequest(PublishingManager.java:152)
maybeSendPublishRequests() (PublishingManager.java:106-112) increments
pendingCount *before* calling sendPublishRequest(), and getTimeoutHint() is
called inside sendPublishRequest(). The throw unwound past the already-taken
permit with no request sent and no completion handler registered, so nothing
ever decremented it. It escaped into the discarded dependent future of the
whenComplete() in maybeSendPublishRequests(), so subscription.create()
returned normally and nothing was logged. Each attempt ratcheted pendingCount
up until it hit maxPendingPublishes, after which Publish traffic stopped for
the remainder of the Session.
The trigger needs no misbehaving Server: a Server may revise the publishing
interval and max keep-alive count freely (Part 4, 5.14.2.2), and nothing bounds
their product.
Fixed by assigning the correct variable and clamping to UInteger.MAX_VALUE
rather than substituting 0, which would disable the Client-side request
timeout entirely and reintroduce a permit leak whenever a Server never
answered. sendPublishRequest() now also wraps request construction and
dispatch in a try/catch that releases the caller's permit and logs before
returning, so any synchronous failure on that path can no longer stall
publishing.
PublishTimeoutHintOverflowTest drives a Server that rewrites its
CreateSubscription response to advertise a 3 600 000 ms revised publishing
interval and a revised max keep-alive count of 1000 (product 1.08e10 after the
in-flight and safety factors, well past UInt32). It asserts that a Publish
request is still sent after the Subscription is created, that
sendPublishRequest() does not throw, and — as a control against a vacuous
result — that ordinary revised parameters also produce a Publish request.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
OpcUaSubscription.modify() took the pending Modifications and cleared the
field before calling the ModifySubscription service:
Modifications diff = modifications;
modifications = null; // cleared BEFORE the service call
assert diff != null;
ModifySubscriptionResponse response = client.modifySubscription(...);
If the service call threw, syncState stayed UNSYNCHRONIZED but the
modifications field was already null, so the caller's requested parameters
were gone. The next modify() re-entered the UNSYNCHRONIZED branch, read a
null diff and blew up at OpcUaSubscription.java:226 with an AssertionError
(surefire enables assertions), or, with assertions disabled, at
OpcUaSubscription.java:231 with a NullPointerException on
diff.publishingInterval(). No concurrency was required to hit this: a single
failed modify() followed by a retry was enough. A transient service failure
is exactly the case a caller is expected to retry, and per OPC UA Part 4
5.13.3 ModifySubscription only revises the parameters the request carries, so
losing the diff means the retry could never apply the requested values.
The modify service call is now wrapped in a try/catch that restores the diff
as the pending Modifications before rethrowing, via the new private
restorePendingModifications(). Modifications requested while the failed call
was in flight take precedence over the restored values, which preserves the
existing behavior of clearing the field before the call so that concurrent
setter calls are not swallowed on success.
SubscriptionModifyRetryTest.modifyAfterFailedModifyRetriesTheSamePendingModifications
uses the DelegatingSubscriptionServiceSet seam to fail the first
ModifySubscription with Bad_TooManyOperations and record every request the
server receives. It asserts the first attempt really did request the new
PublishingInterval, that the Subscription is still UNSYNCHRONIZED after the
failure, and that the second modify() succeeds, reaches SYNCHRONIZED, and
sends the same requested PublishingInterval rather than falling back to the
server's current value.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
OpcUaSubscription's field initializers run before any constructor body, so
maxKeepAliveCount and lifetimeCount were always computed from
DEFAULT_PUBLISHING_INTERVAL (1000 ms):
OpcUaSubscription.java:98-101
publishingInterval = DEFAULT_PUBLISHING_INTERVAL; // 1000.0
maxKeepAliveCount = calculateMaxKeepAliveCount(publishingInterval,
DEFAULT_TARGET_KEEP_ALIVE_INTERVAL); // 10
lifetimeCount = calculateLifetimeCount(maxKeepAliveCount); // 50
The OpcUaSubscription(OpcUaClient, double) constructor
(OpcUaSubscription.java:125-130) then overwrote only publishingInterval, leaving
both counts at their 1000 ms-derived values. create()
(OpcUaSubscription.java:149-156) recomputes them only when they are null, so it
did not rescue the case.
calculateMaxKeepAliveCount is max(1, ceil(target / max(1, interval))) and
calculateLifetimeCount is 5 * maxKeepAliveCount
(OpcUaSubscription.java:1450-1468); with DEFAULT_TARGET_KEEP_ALIVE_INTERVAL of
10000 ms the intended invariant is "a keep-alive roughly every 10 s". Instead,
new OpcUaSubscription(client, 100.0) put RequestedMaxKeepAliveCount=10 and
RequestedLifetimeCount=50 on the wire in CreateSubscription rather than 100 and
500, so:
- the keep-alive cadence became 100 ms x 10 = 1000 ms instead of
100 ms x 100 = 10,000 ms (10x too fast);
- the Subscription lifetime became 100 ms x 50 = 5,000 ms instead of
100 ms x 500 = 50,000 ms (10x too short), so the Server may terminate the
Subscription far sooner than intended (Part 4, 5.14.2.2: the lifetime count
is the number of publishing cycles that may expire without the Server
receiving a Subscription Service request from the Client);
- the client watchdog inherited the same 10x error, because
WatchdogTimer.scheduleNext() (OpcUaSubscription.java:1629-1636) derives its
delay from the revised values the Server echoes back:
publishingInterval x (maxKeepAliveCount + 1) x watchdogMultiplier
= 100 x 11 x 1.5 = 1,650 ms instead of 100 x 101 x 1.5 = 15,150 ms, risking
spurious watchdog trips.
Both errors scale as requestedInterval / 1000, so they get worse the faster the
requested PublishingInterval is.
Fixed by having the two-arg constructor delegate to the one-arg constructor and
then call setPublishingInterval(), which is already the single site that derives
the counts and already honours the lifetimeAndKeepAliveCalculated flag. That
keeps one derivation site instead of duplicating the calculation, which is the
drift that caused the defect. The flag defaults to true and cannot be set to
false before the constructor runs, so no explicitly configured count can be
clobbered, and setPublishingInterval() no-ops on the modifications/syncState
bookkeeping while syncState is INITIAL. setPublishingInterval() itself was
checked and was already correct; this was a single-site defect.
New test SubscriptionKeepAliveDerivationTest captures the CreateSubscription
request the client actually puts on the wire, using a
DelegatingSubscriptionServiceSet subclass registered on every endpoint, and
asserts that new OpcUaSubscription(client, 100.0) requests
PublishingInterval=100, MaxKeepAliveCount=100 and LifetimeCount=500. A second
control test drives the same expectations through the already-correct
setPublishingInterval() path so the expected numbers cannot be vacuous.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Two coupled defects in the client-side notification fan-out, both living in OpcUaSubscription's notify* methods. Listener failure isolation was incomplete. notifyDataReceived and notifyEventsReceived called the application's SubscriptionListener and then each MonitoredItem's listener with no guard, so the first callback to throw abandoned every sink behind it. Worse, the throw escaped PublishingManager.deliverNotificationMessage, whose loop over NotificationMessage.notificationData then skipped every remaining element; the only containment was TaskQueue.TaskWrapper.run()'s catch (Throwable), so publishing carried on while the notification was silently half-delivered. A throwing DataValueListener produced: at OpcUaMonitoredItem.notifyDataValueReceived(OpcUaMonitoredItem.java:486) at OpcUaSubscription.notifyDataReceived(OpcUaSubscription.java:1515) at PublishingManager.deliverNotificationMessage(PublishingManager.java:351) and the StatusChangeNotification that followed the DataChangeNotification in the same NotificationMessage was dropped. The two ArrayLists were also handed to the SubscriptionListener directly and then re-read by the per-item fan-out, so items.clear() suppressed the fan-out entirely and items.remove(k) shifted the items while the values kept their positions, pairing items with the wrong DataValue. Every callback invocation now goes through deliverToListener, which catches and logs any Exception, and the SubscriptionListener receives unmodifiable views of the Lists the fan-out iterates. Callbacks were also delivered out of order. deliverNotificationMessage always runs on the Subscription's delivery queue, but notifyKeepAliveReceived and notifyStatusChanged re-enqueued their callback onto that same queue while the data and event callbacks ran inline. TaskQueue is FIFO with one concurrent task, so the re-enqueued callback landed at the tail: with a keep-alive delivery running and a data delivery already queued, the application observed the data notification before the keep-alive that had preceded it. The extra hop also released backpressure early -- the submitted task's future completed once the callback was merely enqueued, so PublishingManager decremented its pending count and sent another PublishRequest before onKeepAliveReceived had run, contradicting the synchronous-processing contract documented on SubscriptionListener. Both callbacks are now invoked inline, inside the delivery task the pending count waits on, and their javadoc gained the blocking caveat the data and event callbacks already carried. This is the SDK's own delivery contract, not a wire requirement: OPC UA Part 4 does not specify client-side callback ordering. notifyNotificationDataLost and notifyTransferFailed are called from off the delivery queue (PublishingManager's processing queue and the session FSM) and must keep enqueuing onto it; that is now stated in their javadoc. SubscriptionNotificationDeliveryTest scripts Publish responses through ScriptableSubscriptionServiceSet and asserts that a throwing SubscriptionListener still lets every MonitoredItem listener run, that a throwing item listener abandons neither the items behind it nor the remaining NotificationData, that the event path behaves the same, that a listener mutating the notification Lists cannot disturb the item-to-value pairing, that a keep-alive is delivered before a data notification that arrived after it, and that no PublishRequest is sent while onKeepAliveReceived is still running. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
OpcUaSubscription.modify() called resetWatchdogTimer() before assigning the new ServerState built from the ModifySubscription response, so the watchdog was always re-armed from the pre-modify PublishingInterval and MaxKeepAliveCount. The watchdog delay is computed in WatchdogTimer.scheduleNext() as publishingInterval * (maxKeepAliveCount + 1) * watchdogMultiplier, read from getServerState(). At OpcUaSubscription.java:255 (pre-fix) this.serverState was still the object captured at :218, so the timer was armed with stale values and nothing re-armed it afterwards; the only other re-arm is PublishingManager.java:190, which needs a PublishResponse to arrive. Concretely, a Subscription at 100 ms / MaxKeepAliveCount 10 arms for 100 * 11 * 1.5 = 1650 ms. Modifying it to 2000 ms / 10 (correct delay 2000 * 11 * 1.5 = 33000 ms) left it armed for 1650 ms, so onWatchdogTimerElapsed fired roughly 31 seconds before the Server's next keep-alive was even due. The mirror case is equally broken: modifying from a slow interval to a fast one left the watchdog armed for the old, longer delay, delaying detection of a Subscription that had actually gone silent. The revised values are the ones the Server committed to in its ModifySubscription response (OPC UA Part 4 section 5.14.3), so they are what the watchdog must track. Fixed by moving the resetWatchdogTimer() call below the this.serverState assignment, so the timer is scheduled from the revised parameters. SubscriptionWatchdogRearmOnModifyTest covers both directions. Every Publish is parked by ScriptableSubscriptionServiceSet so no PublishResponse can mask the arming done by modify(), and the revised PublishingInterval and MaxKeepAliveCount are asserted before each timing assertion so a Server that revises the requested values fails loudly. modifyToSlowerIntervalDoesNotElapseWatchdogAtThePreModifyInterval asserts the watchdog does not elapse within 6 s after modifying 100 ms -> 2000 ms; modifyToFasterIntervalElapsesWatchdogAtThePostModifyInterval asserts it does elapse within 6 s after modifying 2000 ms -> 100 ms, which also proves the first test is not vacuous. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Two coupled defects in the client Subscription watchdog lifecycle. 1. A Session fault destroyed the watchdog permanently. PublishingManager.sendPublishRequest() handled a Publish failing with Bad_SessionClosed or Bad_SessionIdInvalid by calling OpcUaSubscription.cancelWatchdogTimer() on every Subscription. That method is the teardown path: it de-registers the WatchdogTimer from the client's SessionActivityListeners, cancels the pending expiry, and nulls the watchdogTimer field. The only place a WatchdogTimer is ever constructed is OpcUaSubscription.create(), which is guarded by syncState == INITIAL and throws Bad_InvalidState otherwise, so a still-SYNCHRONIZED Subscription can never get another one. Both status codes are classified as Session errors by SessionFsmFactory.SessionFaultListener, which drives the FSM through re-activation; TransferSubscriptions then keeps the Subscription alive on the Server and Publish traffic resumes, because PublishingManager's own onSessionActive listener is still registered. Only the watchdog is lost. The concrete failure was therefore: transient Session fault -> watchdog cancelled and de-registered -> Session re-activates -> TransferSubscriptions succeeds -> Subscription runs unsupervised for the rest of its life, so a Server that later stops honouring the keep-alive interval it promised in its CreateSubscription response (Part 4, 5.13.2) is never reported to the application. Fixed by adding OpcUaSubscription.pauseWatchdogTimer(), which cancels the pending expiry but leaves the WatchdogTimer registered and reachable, and calling that from the Session-fault path. WatchdogTimer.onSessionActive() re-arms it when the Session comes back. cancelWatchdogTimer() is unchanged and still used by reset() for genuine Subscription teardown; the distinction now encoded is temporary Session unavailability versus permanent removal. WatchdogTimer.onSessionInactive() likewise now pauses instead of cancelling, which is what it always meant. 2. The timer's state transitions were not atomic. WatchdogTimer held its pending expiry in an AtomicReference and performed a read/cancel/schedule/store sequence across it. reset() also used get() rather than getAndSet(). The outer wrappers resetWatchdogTimer() and cancelWatchdogTimer() are synchronized on the OpcUaSubscription, but the Session callbacks bypassed them and called the inner reset()/cancel() directly, so the two entry points shared no lock at all. They run concurrently in practice: SessionFsmFactory dispatches activity listeners on the transport executor, and PublishingManager completes Publish responses with whenCompleteAsync on that same executor before calling resetWatchdogTimer(). It defaults to Stack.sharedExecutor(), an unbounded cached pool. The losing interleaving: T1 in reset() reads sf0, cancels it, and enters scheduleNext(), which creates sf1; T2 in cancel() does getAndSet(null) -> sf0 and re-cancels the already-cancelled sf0; T1 then stores sf1. sf1 is live after a cancel that happened after it was created, and it fires, telling the application a healthy Subscription is dead. Via cancelWatchdogTimer() the WatchdogTimer was also unreachable by then, so nothing could ever cancel sf1. Independently, two concurrent reset() calls both cancelled sf0 and both scheduled, leaking whichever future lost the store. Fixed by making WatchdogTimer self-synchronized on a private lock covering every transition, with a terminal cancelled flag so a cancelled timer never arms again, and an epoch captured when each expiry is scheduled and re-checked inside notifyWatchdogTimerElapsed(). The epoch closes the remaining window where an expiry has already started running and cancel(false) cannot stop it. Nothing under the new lock reaches back for the OpcUaSubscription monitor - getServerState() and watchdogMultiplier are volatile reads - so the lock order remains one-way. New tests: SubscriptionWatchdogSessionFaultTest answers the first Publish with Bad_SessionIdInvalid and parks every subsequent one, so no Publish response can re-arm the watchdog. It asserts the fault took the Session out of Active and that it was re-activated, then asserts onWatchdogTimerElapsed does fire after re-activation, counting only expiries observed after the return to Active. Its control asserts the watchdog fires in the same fixture with no fault at all. SubscriptionWatchdogCancelRaceTest forces the interleaving instead of racing for it: the client is given a ScheduledExecutorService that suspends the re-arming thread inside schedule(), after the current future was cancelled and before the new one is stored, and releases onSessionInactive() into that window. It asserts a cancelled watchdog does not fire. Two controls bound it: the same gated re-arm with no cancel does fire, and the same two operations performed sequentially do cancel. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Sequence-number accounting in PublishingManager.processPublishResponse had three defects that let NotificationMessages be lost silently. They share one root cause -- what the sequence number of a keep-alive means -- and one fix, and cannot be separated: the "receivedSequenceNumber == 1" disjunct that caused the first defect was also the only thing that ever un-stuck the state after a rollover, so removing it requires the modular arithmetic to land with it. WHAT THE DEFECTS WERE (line numbers are from the parent commit) 1. An initial keep-alive masked the loss of NotificationMessage 1. PublishingManager.java:300 set lastSequenceNumber for any message numbered 1, keep-alive or not. Part 4 §5.14.1.1 says a keep-alive "does not contain any Notifications and ... contains the sequence number of the next NotificationMessage that is to be sent", so a keep-alive carrying 1 means message 1 has *not* been sent. Trace: keep-alive(1) set lastSequenceNumber = 1; NotificationMessage 1 was then sent and lost; message 2 arrived with expected == 2, so "received > expected" was false, no gap was detected, and message 1 was never requested via Republish. Aggravating it, the block at PublishingManager.java:307-314 refilled the acknowledgement list verbatim from availableSequenceNumbers, so the client acknowledged sequence 1 -- a message it never received. Part 4 §5.14.5.2: "The Server may delete the Message with this sequence number from its retransmission queue", so the acknowledgement destroyed the only remaining copy. 2. A gap ending in a keep-alive was recovered repeatedly. The loop bounds at PublishingManager.java:271-273 were right, but PublishingManager.java:297 then advanced lastSequenceNumber to the *first* missing sequence number instead of the last, and the "== 1 || !isKeepAlive" guard declined to correct it on the keep-alive path. With lastSequenceNumber = 5 and a keep-alive carrying 10, the first pass republished 6,7,8,9 and left lastSequenceNumber = 6, so the next keep-alive republished 7,8,9, then 8,9, then 9: O(n^2) blocking Republish calls for an n-message gap. It terminated, but the first pass had already acknowledged 6-9, so the Server had purged them and every later cycle failed, raising up to n-1 spurious notifyNotificationDataLost() callbacks -- or, if the acknowledgements had not landed yet, delivering the same notifications to the application twice. 3. None of the arithmetic wrapped. PublishingManager.java:257 computed "lastSequenceNumber + 1", which yields 4294967296 -- a value no NotificationMessage can carry -- and PublishingManager.java:268 compared with a plain ">". With lastSequenceNumber = 0xfffffffe and message 0xffffffff lost, the wrapped message 1 compared as behind and the gap vanished; with lastSequenceNumber = 0xffffffff, no sequence number could ever again compare as ahead and gap detection was dead for the life of the Subscription. Note that delivery is unconditional either way: what these defects corrupt is the gap-detection state, and the acknowledgement block then acknowledged messages on the strength of it. HOW IT WAS FIXED New package-private SequenceNumbers holds the arithmetic. Part 4 §5.14.1.1 -- "The value 0 is never used for the sequence number. The first NotificationMessage sent on a Subscription has a sequence number of 1. If the sequence number rolls over, it rolls over to 1" -- makes sequence numbers a cycle over 1..0xFFFFFFFF, so successor(0xFFFFFFFF) is 1 and predecessor(1) is 0xFFFFFFFF. This is deliberately *not* the plain Counter of Part 4 §7.8, whose successor of 0xFFFFFFFF is 0; using §7.8 here would be wrong by one at every rollover. "Nothing accounted for yet" is represented by keeping the existing lastSequenceNumber = 0 sentinel, which §5.14.1.1 guarantees is not a legal sequence number. successor(0) is defined as 1, the sequence number of the first NotificationMessage a Subscription sends, so the initial state needs no special case; predecessor(0) and every other operation reject it. A NotificationMessage carrying an illegal sequence number is now logged and delivered, but leaves the accounting untouched rather than feeding a nonsensical value into it. processPublishResponse now applies one rule for both defects 1 and 2: lastSequenceNumber = isKeepAlive ? predecessor(received) : received. A keep-alive accounts for everything through the predecessor of the sequence number it carries and is not evidence that that sequence number arrived. Acknowledgements are no longer refilled from availableSequenceNumbers. The client now acknowledges exactly the NotificationMessages it took delivery of: the received message when it is not a keep-alive, plus each message recovered by Republish. Acknowledging a message that never arrived is what turns a detected gap into permanent data loss (§5.14.5.2). Recovery is now bounded. With modular distance, a stale or corrupt sequence number can be up to 2^31 steps "ahead", and the recovery loop issues one synchronous Republish per missing message on the shared processing queue. The gap is therefore capped at the number of messages the Server says it is holding in availableSequenceNumbers, falling back to 64 when it advertises none; a larger gap is logged, reported through notifyNotificationDataLost(), and resynchronized to the received sequence number instead of iterated. Republish recovery remains synchronous; making it non-blocking is a separate change. ON THE #1401 INTEROP CASE Commit 33438b7 added the "== 1" disjunct to avoid "an unnecessary Republish call when a Server sends an empty (keep-alive) first PublishResponse followed by a second PublishResponse once it has acquired initial values". Milo's own Server sends keep-alive(1) then data(1) -- its keep-alive reports the next sequence number without consuming it -- and that trace is unchanged here: expected == 1, received == 1, no gap. The trace the disjunct actually suppressed is keep-alive(1) followed by data(2), and per §5.14.1.1 that means message 1 was sent and did not arrive, so Republish(1) is correct rather than doomed. Against a Server that instead consumes a sequence number for its keep-alives, this costs one failed Republish and one notifyNotificationDataLost() per Subscription; that is the honest report, since the client cannot account for the message and has no way to tell that case apart from a genuine loss. WHAT THE NEW TESTS ASSERT SequenceNumbersTest (sdk-client, no Server) pins the arithmetic down at the boundary the wire cannot reach: successor(0xFFFFFFFF) == 1, predecessor(1) == 0xFFFFFFFF, that no operation ever produces 0, that 1 is "ahead" of an expected 0xFFFFFFFF, that a sequence number more than half a cycle ahead is treated as stale rather than as an enormous gap, and that walking a gap from 0xFFFFFFFF to 2 enumerates exactly 0xFFFFFFFF and 1. PublishSequenceRecoveryTest (integration-tests) drives real PublishResponses through the real client stack with a scripted Server and asserts: a NotificationMessage lost after an initial keep-alive is requested via Republish and is never acknowledged; each message missing before a keep-alive is requested exactly once and a fully recovered gap reports no lost data; and, seeding lastSequenceNumber reflectively because the rollover is unreachable over the wire, that a message lost at the rollover boundary is still detected. A positive control covers a gap between two data messages, which was already detected before this commit. TestServer's free-port probe bound InetAddress.getLocalHost(), which resolves to 127.0.1.1 on some hosts while the Server binds 127.0.0.1, so an occupied port could pass the probe and then fail the real bind with Bad_ConfigurationError. It now probes the wildcard address. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
The transport goes out of its way to preserve the order the Server sent
NotificationMessages in: AbstractUascClientTransport.handleResponse:146-147
completes PublishResponse futures on a serial ExecutionQueue, whose
concurrency limit of 1 guarantees "tasks run serially and in the order
submitted". PublishingManager then handed its completion handler to
whenCompleteAsync(..., client.getTransport().getConfig().getExecutor()),
which is by default Stack.sharedExecutor() -- an unbounded cached pool.
The dispatches happened in wire order on the ExecutionQueue thread, but
the handlers ran on arbitrary pool threads, so any two of them could
invert before reaching processingQueue.execute().
With lastSequenceNumber == 1 and responses 2 and 3 inverted:
- 3 is processed first. recoverMissingNotificationMessages sees a gap
at 2 and issues a Republish for a message the Server already sent
and the transport already delivered; the recovered copy is delivered
to the application. lastSequenceNumber becomes 3.
- 2 is processed next. It is neither expected nor ahead, so no gap is
detected, but PublishingManager.java:285 assigned lastSequenceNumber
unconditionally and rolled it back from 3 to 2, and the message was
delivered a second time.
- 4 then looks like the far end of a gap at 3, producing a second
spurious Republish -- and the regression repeats for every message
that follows, one blocking Republish round trip each.
The same corruption is reachable with no threading involved at all: any
NotificationMessage that arrives after it has already been accounted
for -- a duplicate, or a copy of a message the client had already
recovered via Republish -- rolled the accounting backwards in exactly
the same way.
Both halves are fixed.
Ordering: the completion handler is registered with whenComplete again,
so for a PublishResponse it runs inline on the transport's serial
PublishResponse queue and hands the work straight to processingQueue,
which is itself serial. Wire order therefore survives into processing.
The handler does nothing that can block: only a map lookup, a watchdog
timer reset (a cancel plus a re-schedule, deliberately kept at the point
of receipt because the watchdog exists to detect the Server going
quiet), and the enqueue. The failure branch is moved to
handlePublishFailure and is still dispatched onto the executor, because
it can run on a wheel timer thread or inline on the caller's thread and
it re-enters maybeSendPublishRequests(); that async hop is what 88dd127
("Use whenCompleteAsync when handling PublishResponse") was protecting,
and it is retained.
Idempotence: processPublishResponse now compares the received sequence
number against the expected one using the modular helpers in
SequenceNumbers. A message that is neither expected nor ahead of
expected has already been accounted for, so it is acknowledged -- it was
genuinely received, and Part 4 5.14.5.2 makes an acknowledgement a
statement about receipt -- and then discarded: no gap detection, no
delivery, and lastSequenceNumber is left alone. Its pending-publish
permit is released so the Publish pipeline does not stall, via the new
releasePendingPublish helper that also replaces the two open-coded
copies of that decrement-and-refill pair.
PublishResponseOrderingTest covers both halves. ReorderedInsideTheClient
injects an executor that deterministically inverts two dispatches from a
single ordered source and asserts that Republish is never invoked and
that each NotificationMessage reaches the application exactly once.
StaleNotificationMessage scripts the Server to send sequence 3 before 2
and asserts that the correct Republish for 2 is the only one, and again
that nothing is delivered twice. Two controls guard the fixture: a
genuine gap must still be recorded by the Republish responder, and the
injecting executor must really invert. ScriptableSubscriptionServiceSet
gains getParkedRequestCount() so a test can wait for the client's
Publish pipeline to fill rather than sleep.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Republish recovery ran on the queue that processes PublishResponses and
blocked it for a network round trip per missing NotificationMessage --
on a queue shared by every Subscription on the client:
- PublishingManager.java:78 built one processingQueue for the whole
client from the transport executor. TaskQueue's default concurrency
limit is 1, so PublishResponses for every Subscription were
processed serially on it.
- PublishingManager.java:434, on that queue, called
OpcUaClient.republish(), which is republishAsync(...).get()
(OpcUaClient.java:2621-2628): an unbounded blocking wait.
- AbstractUascClientTransport.handleResponse completes anything that
is not a PublishResponse via config.getExecutor().execute(...)
(AbstractUascClientTransport.java:149) -- the same executor the
processing queue runs on.
With a single-threaded executor -- an ordinary application choice,
OpcTcpClientTransportConfigBuilder.setExecutor exists for it -- the only
thread that could complete the RepublishResponse was the one waiting for
it, and the deadlock was permanent rather than merely slow:
handleResponse cancels the request timeout
(AbstractUascClientTransport.java:144) before queueing the completion,
so the wheel timer that rescues a Server that never answers cannot
rescue a Server that does. With threads to spare there was no deadlock,
but each missing message still stalled the processing of PublishResponses
for every other Subscription for a full round trip.
Recovery also spent those round trips on NotificationMessages the Server
had just said it no longer holds: it walked the numeric gap and ignored
the availableSequenceNumbers of the response that revealed it.
Recovery is now an asynchronous, per-Subscription state machine.
Each Subscription owns its processing queue, so one Subscription's
recovery no longer holds up another's PublishResponses. The ordering
established by d52122e is preserved: the completion handler still
enqueues inline on the transport's serial PublishResponse queue, and
each per-Subscription queue is serial, so wire order still survives into
processing. While a gap is being recovered that Subscription's queue is
paused rather than blocked, which keeps the recovered NotificationMessages
ahead of the one that revealed the gap and stops a later PublishResponse
overtaking the recovery and rolling the sequence accounting forward
under it.
The Republish requests are chained one at a time through
republishAsync(), and each recovered NotificationMessage is handed to the
delivery queue as it arrives; nothing waits for a round trip. Part 4
5.14.1.1: "In the case of a retransmission queue overflow, the oldest
sent NotificationMessage gets deleted", so a missing sequence number
older than the oldest the Server advertises in availableSequenceNumbers
is gone for good and is reported as lost data instead of costing a round
trip that can only be answered Bad_MessageNotAvailable. A Republish that
does fail with Bad_MessageNotAvailable no longer ends the pass -- the
queue is purged oldest-first, so the messages after it may well still be
there -- but any other failure is the service call itself failing and
abandons the sequence numbers not yet requested rather than repeating a
request that can only fail the same way. The bound on how large a gap is
worth trying to recover, added in 3fbb57c, is unchanged.
RepublishRecoveryTest covers all three defects:
- BlockedByItsOwnRecovery drives a gap of three NotificationMessages
through a client whose transport executor has one thread, against a
Server that answers every Republish immediately, and asserts that all
three are requested and that the message that revealed the gap is
delivered. Its control runs the identical script on a cached pool, so
the executor is isolated as the cause.
- RecoveryOnTheSharedProcessingQueue holds a Republish response open and
asserts that a PublishResponse for a second, healthy Subscription is
still processed while that recovery is in flight.
- AvailableSequenceNumbers asserts that a Server advertising 3, 4 and 5
is asked to retransmit 3 and 4 only, and -- as the control against a
fix that simply stops Republishing -- that one advertising 2, 3, 4
and 5 is asked for all three missing messages.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Two defects in how PublishingManager handles SubscriptionAcknowledgements once it has queued them. Line numbers are from the parent commit, b6faed4. WHAT THE DEFECTS WERE 1. An acknowledgement was abandoned permanently if the PublishRequest carrying it failed. PublishingManager.java:143-162 drained every queued sequence number into the outgoing request and cleared the queue (PublishingManager.java:160). Nothing ever put them back: the failure path, PublishingManager.java:234 -> handlePublishFailure at PublishingManager.java:254, released the pending-publish permit and re-armed the pipeline without them, and the synchronous failure path, PublishingManager.java:237, did the same. Trace: NotificationMessage 1 arrives, PublishingManager.java:323 queues an acknowledgement for it, the next PublishRequest drains it, and that request fails -- a timeout, a dropped connection, a transient ServiceFault. The client has now decided, silently, never to acknowledge sequence 1. Part 4 §5.14.7.1: "The Client should acknowledge all Messages in this list for which it will not request retransmission." The Server is never told, so it keeps message 1 in its retransmission queue and re-advertises it in availableSequenceNumbers on every subsequent PublishResponse until the queue evicts it. Not data loss, but an unbounded-until-eviction leak on the Server and a growing availableSequenceNumbers on every response. This was a regression introduced by 3fbb57c. Before it, the ack block refilled the queue wholesale from the response's availableSequenceNumbers, which incidentally re-queued anything lost this way. 3fbb57c correctly removed that refill -- availableSequenceNumbers is diagnostic data, not proof of receipt, and copying it acknowledged messages the client never received -- but it removed the resync without replacing it. 2. The PublishResponse results array was never read. Part 4 §5.14.5.2 defines it as the "List of results for the acknowledgements", whose "size and order ... matches the size and order of the subscriptionAcknowledgements request parameter". It is how a Server reports Bad_SequenceNumberUnknown or Bad_SubscriptionIdInvalid for one acknowledgement while the Publish call itself succeeds. PublishingManager had no reference to getResults() at all, so an acknowledgement the Server refused was indistinguishable from one it accepted, and nothing was logged. HOW IT WAS FIXED sendPublishRequest now records what it drained, per Subscription, in a list of DrainedAcknowledgements, and restoreAcknowledgements() puts them back on both failure paths: at the top of handlePublishFailure, before the permit is released and before maybeSendPublishRequests() runs, so the next PublishRequest carries them; and in the synchronous catch block alongside the existing permit release. Restored sequence numbers go back at the head of the queue, ahead of anything queued since, because they are older, and a sequence number already queued again is not added twice. The invariant 3fbb57c established is preserved: the only things that ever enter the queue are the received message when it is not a keep-alive (PublishingManager.java:323) and each message recovered by Republish (PublishingManager.java:655). Nothing is copied from availableSequenceNumbers. Restoring only moves an entry back to where it was. Re-queuing is safe in the weaker failure case, where the request did reach the Server and only the response was lost: a repeated acknowledgement is answered with a per-acknowledgement result, not a fault. Milo's own Server does exactly that -- SubscriptionManager.java:1416 calls Subscription.acknowledge(), which returns Bad_SequenceNumberUnknown (Subscription.java:802) when the message is no longer in availableMessages -- and that result is now logged rather than silently dropped. The drain loop also no longer discards acknowledgements for a Subscription whose subscriptionId is absent. The old code cleared the queue inside the lambda whether or not ifPresent() had produced anything, which is the same defect in miniature. reportRefusedAcknowledgements() pairs response.getResults() positionally with the acknowledgements the request carried and logs each bad result at WARN with its subscriptionId and sequenceNumber. A length mismatch is logged at DEBUG and only the common prefix is paired, since a Server that returns a short or empty results array is not something to warn about on every response. A refused acknowledgement is reported and then forgotten, never re-queued. Both codes a Server uses here are statements that there is nothing left to acknowledge -- Bad_SequenceNumberUnknown means it is not holding a NotificationMessage with that sequence number, Bad_SubscriptionIdInvalid that the Subscription is gone -- so re-queuing could only be refused the same way, forever, while keeping the entry alive in the client's queue indefinitely. WHAT THE TESTS ASSERT PublishAcknowledgementTest (integration-tests) drives a real client against a scripted Server. AcknowledgementLostWithItsPublishRequest arms a service set that fails one PublishRequest -- the one carrying the acknowledgement for sequence 1 -- with Bad_Timeout before super.onPublish records anything, so the model is the strong one: the request and its acknowledgement were genuinely lost in flight and the Server never saw them, which keeps getReceivedAcknowledgements() a sound record of what the client managed to acknowledge. It then asserts the acknowledgement for sequence 1 does eventually reach the Server. Its sibling is the control: the identical script with nothing failed, so the assertion cannot pass vacuously. AcknowledgementResults scripts a PublishResponse whose results array refuses the acknowledgement for sequence 1 with Bad_SequenceNumberUnknown, positionally, and asserts the client reports it. There is no listener, status, or other API surface on which a refused acknowledgement could be observed, so it asserts on log output, with a WARN probe through the same SLF4J binding as a control that the capture is live. That makes it a weaker test than the others, and it constrains the diagnostic to WARN or above; it is the best available without adding API surface for it. The synchronous restore, in sendPublishRequest's catch block, has no seam to drive from a test and is covered by inspection only. ScriptableSubscriptionServiceSet gains a buildPublishResponse overload taking an explicit StatusCode[] results; the existing method delegates to it with a GOOD-filled array, so no existing script changes behavior. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Two defects in PublishingManager#handlePublishFailure, both about when the
Publish pipeline is refilled once a PublishRequest comes back as a fault
rather than a PublishResponse.
Bad_NoSubscription could strand the pipeline for good
-----------------------------------------------------
Part 4 §5.14.8.1: when the last Subscription of a Session is deleted, "all
Publish requests still queued for that Session are de-queued and shall be
returned with Bad_NoSubscription". Milo's own Server does exactly that in
SubscriptionManager.
An application that deletes its last Subscription and immediately creates
another therefore takes that whole de-queued burst *after* its Subscription
set is non-empty again, and the client had nothing left that would send a
PublishRequest for the new Subscription:
- OpcUaSubscription#delete() calls deleteSubscriptions and then reset(),
which calls removeSubscription(); subscriptionDetails is now empty, so
getMaxPendingPublishes() returns 0 and that refill is a no-op. The two
permits taken by the in-flight requests are still held.
- create() calls addSubscription(), whose refill loop
(PublishingManager.java:118, "for (long i = pendingCount.get(); i <
maxPendingPublishes; i++)") runs with pendingCount already at 2 and a
target of min(1 + 1, maxPendingPublishRequests) = 2, so it sends nothing.
- the de-queued requests then fail with Bad_NoSubscription, and
handlePublishFailure (PublishingManager.java:391) decremented the count
but deliberately skipped maybeSendPublishRequests() for that status code.
With pendingCount driven to 0 and no request in flight, none of the remaining
callers of maybeSendPublishRequests() can fire again, and the watchdog only
reports the silence — notifyWatchdogTimerElapsed() invokes the listener and
performs no recovery. The Subscription is created, the Server publishes for
it, and the application never receives anything.
The suppression itself is worth keeping: replacing a request the Server
answered Bad_NoSubscription with another one for the same Subscription set
could only be answered the same way, in a hot loop. What was missing is that
the answer describes the Subscription set the request was sent for. A
subscriptionGeneration counter, bumped whenever a Subscription is registered
or unregistered and recorded by each PublishRequest as it is sent, now tells
the two apart: the refill is suppressed only while the set is unchanged. The
refilled requests carry the new generation, so a Server that keeps answering
Bad_NoSubscription still cannot be hammered.
Bad_TooManyPublishRequests refilled the whole deficit at once
------------------------------------------------------------
Part 4 §5.14.5.1: after this error a Client "shall not issue another Publish
request before one of its outstanding Publish requests is returned". Not
replacing the failed request on the spot already satisfied the letter of
that, but the target stayed at subscriptionCount + 1, so the deficit the
fault opened was still there and the refill loop closed all of it the moment
the next PublishResponse was delivered. With one Subscription: the fault
takes pendingCount 2 -> 1, the next delivered response takes it 1 -> 0, and
two new requests go out — one request returning, two issued, restoring
exactly the outstanding count that had just drawn the fault. The result is a
steady-state oscillation of one extra request and one ServiceFault per
delivery; the pipeline does not stall and no data is lost.
handlePublishFailure now also learns a ceiling from the fault — the number of
requests the Server was in fact holding, i.e. the outstanding count after the
failed one is accounted for, floored at one so the ceiling can never mean
"never send another PublishRequest" — and getMaxPendingPublishes() applies it
on top of the existing target. Recovery is deliberately conservative rather
than timed: the ceiling is cleared when a Subscription is added, because the
client now wants a deeper pipeline than when the ceiling was learned, and on
Session activation, because a new Session has its own queue.
Note that the same clause requires a Server to accept at least
subscriptionCount + 1 queued Publish requests, which is precisely what the
client aims for, so a conformant Server never answers
Bad_TooManyPublishRequests here at all. This matters only against a Server
with an off-by-one or a fixed cap well below the Subscription count, which is
also why the fix is scoped to a learned ceiling rather than a redesign of the
pipelining.
Tests
-----
PublishPipelineRefillTest drives both through ScriptableSubscriptionServiceSet,
which parks every Publish request until the test decides how it ends, so the
timing belongs to the test rather than to a Server timer.
DeletingTheLastSubscription fills the pipeline with two parked requests,
deletes the only Subscription, creates another, then fails the parked
requests with Bad_NoSubscription and asserts that Publish requests are sent
again — once mechanically, and once as what the application sees, by asserting
the re-created Subscription actually receives a NotificationMessage. A control
runs the identical sequence with Bad_InternalError, which the failure handler
never suppressed, proving the fixture can observe a refill at all.
TooManyPublishRequests faults one of two parked requests, asserts no
replacement is sent for it (the control, which passed before this change and
is also the barrier that makes the failure accounted for before the next
response is delivered), then delivers a keep-alive on the survivor and asserts
that exactly one replacement follows rather than two.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Two defects in the pending-deletion queue OpcUaSubscription keeps for
MonitoredItems that have been removed from the Subscription but not yet
deleted from the Server. They are coupled — retaining failed deletions makes
the second defect strictly more likely to fire — so both are fixed together.
A failed DeleteMonitoredItems dropped the pending deletion
----------------------------------------------------------
deleteMonitoredItems() (OpcUaSubscription.java:672-685) cleared
this.itemsToDelete *before* the service call. The private overload can then
fail three ways without ever reaching applyDeleteResult(): serverState == null
(OpcUaSubscription.java:694), no resolvable MonitoredItemIds
(OpcUaSubscription.java:722), and the service fault catch
(OpcUaSubscription.java:760). In each case the item kept its serverState,
ClientHandle and SyncState.SYNCHRONIZED while being in neither monitoredItems
— removeMonitoredItem() (OpcUaSubscription.java:432) had already taken it out
— nor itemsToDelete. Fully orphaned.
The consequences compound. The Server-side item keeps publishing, and the
notification handler resolves its ClientHandle to nothing. Every queued
deletion having been dropped, isMonitoredItemsSynchronized()
(OpcUaSubscription.java:998) reports true, so the failure is invisible to
callers that poll sync state instead of inspecting the returned results. And
the item cannot be added back: addMonitoredItem()
(OpcUaSubscription.java:386) sees a ClientHandle whose entry in monitoredItems
is not this item, itemsToDelete.remove(item) returns false, and the item falls
into the "handle from a different context, ignore" branch and is silently
dropped. Short of deleting the whole Subscription there was no way for an
application to recover.
deleteMonitoredItems() now snapshots the queue rather than clearing it, and
afterwards removes only the items for which the Server reported an
operation-level result. Part 4 §5.13.6.4, Table 77 defines exactly one
operation-level result code for DeleteMonitoredItems besides Good:
Bad_MonitoredItemIdInvalid, which means the item is already gone. Either way
the item no longer exists on the Server and applyDeleteResult() has detached
it, so dequeuing is correct. Anything without an operation-level result — a
service fault, a missing serverState, an unresolvable MonitoredItemId — leaves
the item on the Server, so it stays queued for the next attempt. Items still
in SyncState.INITIAL were never created on the Server and are still dropped
without a service call.
reset() replayed stale MonitoredItemIds against the new Subscription
--------------------------------------------------------------------
reset() (OpcUaSubscription.java:1424) reset monitoredItems.values() but never
touched itemsToDelete. OpcUaMonitoredItem.reset() is what clears serverState,
and therefore the MonitoredItemId, so items sitting in the queue kept both a
stale MonitoredItemId and SyncState.SYNCHRONIZED across the reset. After a
subsequent create(), deleteMonitoredItems() still selected them (SyncState !=
INITIAL) and sent those ids against the new subscription id.
MonitoredItemIds are scoped to a Subscription, and Milo's own Server assigns
them from an AtomicLong created per Subscription and starting at 1
(sdk-server Subscription.java:75), so the ids restart for every new
Subscription. A replayed id of 1 therefore usually maps to a real, unrelated
item in the recreated Subscription: the delete does not fail, it silently
removes the wrong item. Separately, isMonitoredItemsSynchronized() stayed
false indefinitely while the ghosts remained queued.
reset() now resets the queued items and clears the queue: the Subscription
they belonged to no longer exists, so neither do they, and there is nothing
left to delete. Their ClientHandle is cleared too, so an item that outlives
its Subscription can be added to one again instead of hitting the same
"handle from a different context, ignore" branch as the first defect.
Tests
-----
MonitoredItemDeletionTest drives a real Server whose MonitoredItem service set
is wrapped to record the MonitoredItemIds of every DeleteMonitoredItems
request and to fail scripted ones with a service fault:
- failedDeleteStaysQueuedForTheNextDelete: after a service fault the
Subscription reports itself unsynchronized, the next
deleteMonitoredItems() retries the same item, and the Server sees the
deletion attempted twice.
- itemWhoseDeleteFailedCanBeAddedBack: an item whose delete failed is
restored to the Subscription, still SYNCHRONIZED, when added back.
- staleMonitoredItemIdIsNotDeletedFromTheNewSubscription: pins the collision
by asserting both the old and the new item are assigned MonitoredItemId 1,
then asserts no DeleteMonitoredItems reaches the wire after the
Subscription is recreated and that the item holding the recycled id can
still be deleted itself.
- newSubscriptionIsSynchronizedAfterAnItemWasQueuedForDeletion: the
recreated Subscription reports itself synchronized.
Three controls guard against vacuous fixes: a Good delete must still dequeue
(so "never dequeue" is not a valid fix), an item added back while queued must
cancel its pending deletion, and a delete/create cycle with no pending
deletion must still require its items to be created again.
DelegatingMonitoredItemServiceSet is new test infrastructure mirroring the
existing DelegatingSubscriptionServiceSet; there was no delegating
MonitoredItem service set.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
An OpcUaSubscription object and the Subscription it represents did not have the same lifetime, and nothing in the client said so. reset() cleared the object's ServerState and create() installed a new one, while every piece of PublishingManager work registered for the previous Subscription kept asking that same mutable object what its *current* SubscriptionId was. Work belonging to one Subscription was therefore applied to another. The defects 1. Stale work from a previous incarnation. SubscriptionDetails carried only the OpcUaSubscription object. The Bad_Timeout branch of deliverNotificationMessage (PublishingManager.java:897-909) asked the live object for its id and removed *that* map entry, then called notifyStatusChanged, which resets the live object (OpcUaSubscription.java:1634-1643). Delivery runs on the Subscription's deliveryQueue, a final field (OpcUaSubscription.java:114) that reset() neither drains nor replaces (OpcUaSubscription.java:1443-1467), so a task queued behind an application callback that has not returned survives into the next incarnation. Traced outcome: a Bad_Timeout StatusChangeNotification received for Subscription A de-registers and tears down Subscription B, created after it and never timed out, leaving B running on the Server with no ServerState left for delete() to name it. The same interleaving with a DataChangeNotification delivered a value received on A to the application as though a MonitoredItem of B had reported it: reset() leaves the monitoredItems map and its ClientHandles intact, so the stale value still resolves to an item. 2. Acknowledgements built from the wrong SubscriptionId. The acknowledgement loop iterated subscriptionDetails.values() and built each SubscriptionAcknowledgement from the live object's current id (PublishingManager.java:220) rather than the id its entry was registered under. Part 4 section 5.14.5.2 pairs a sequenceNumber with the Subscription the NotificationMessage was "received on" and lets the Server delete that message from that Subscription's retransmission queue, so the client both acknowledged a message the named Subscription never sent and left the one it did receive unacknowledged in the Server's queue. 3. A leaked Server-side Subscription. create() was a check of syncState around a blocking CreateSubscription call with nothing serializing it (OpcUaSubscription.java:157-197), so two concurrent calls both passed the check and both created a Subscription. The object kept only the ServerState written last; the other SubscriptionId could never be passed to delete() and the Subscription ran on the Server until its lifetime expired. create() also published syncState = SYNCHRONIZED before serverState, so getSyncState() could report SYNCHRONIZED while getSubscriptionId() was still empty. The fix SubscriptionDetails now holds the SubscriptionId it was registered under as a final field, plus a registered flag that is cleared when the entry is unregistered. Acknowledgements are built from that id. The Bad_Timeout branch unregisters the entry by its own id. deliverNotificationMessage drops a message whose entry is no longer registered, which covers data, event, keep-alive and status delivery in one place; the entry is unregistered the moment its Subscription is, so a live Subscription's own work is never affected. removeSubscription now unregisters by object identity rather than by the object's current id. OpcUaSubscription serializes create/modify/delete/setPublishingMode/reset on a private lifecycle lock held across the service call, so each check-then-act is atomic: a second concurrent create() is answered Bad_InvalidState instead of creating a Subscription nothing can name, and addSubscription always registers under the id create() just installed. create() now writes serverState before syncState. Both classes gained a class Javadoc stating the threading contract, which previously existed nowhere. The tests SubscriptionIdentityTest drives both ways an entry's id and its Subscription's current id can disagree. StaleWorkFromAPreviousIncarnation blocks the delivery queue inside onDataReceived, queues a second NotificationMessage behind it, resets and recreates the Subscription, then releases the queue: it asserts the recreated Subscription is neither reset nor de-registered by the stale Bad_Timeout, and that the stale DataValue is not delivered. ConcurrentCreate holds two CreateSubscription requests inside the Server handler and releases them in a controlled order: it asserts an acknowledgement carries the id its NotificationMessage was received under, and that delete() leaves no Subscription running on the Server. Five controls fix the assertions in place - that reset()+create() alone leaves a usable Subscription, that a Bad_Timeout does reset the Subscription it was received for, that a queued DataChangeNotification is otherwise delivered, and that acknowledgement and delete work after a single create(). With create() serialized, ConcurrentCreate's second call is now rejected client-side, so its acknowledgement test no longer exercises a divergent id; the invariant it asserts is defended by the immutable id, which the StaleWorkFromAPreviousIncarnation tests still exercise. The syncState-before-serverState ordering has no test: the two statements are adjacent with no interposable call between them and every reader of serverState null-checks it, so it is verified by code order only. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Part 4 §6.7 tells a Client what to do with a Subscription once it has re-established a connection: "After re-establishing the connection the Client shall call Republish in a loop, starting with the next expected sequence number and incrementing the sequence number until the Server returns the status Bad_MessageNotAvailable. After the Republish returns Bad_MessageNotAvailable the Client shall start sending Publish requests with the normal Publish handling. This sequence ensures that the lost NotificationMessages queued in the Server are not overwritten by new Publish responses." The client already did the rest of §6.7 correctly -- it re-establishes the SecureChannel, re-activates or re-creates the Session, and calls TransferSubscriptions with sendInitialValues -- and omitted only that drain. The defects 1. Publish resumed before Republish, on both reconnect paths. PublishingManager's SessionActivityListener went straight from onSessionActive to maybeSendPublishRequests (PublishingManager.java:125-135). Outside the public republish* API the only Republish call site in the client SDK was PublishingManager.java:868-879, reached only from processPublishResponse's reactive gap repair, i.e. after a PublishResponse had already come back and revealed a gap. Publish resumption is gated on State.Active (SessionFsmFactory.java:664-717), and both paths -- Creating -> Activating -> Transferring -> Initializing -> Active, and Active -> ReactivatingWait -> Reactivating -> Initializing -> Active -- reach it with no Republish in between, so the first request the client sent for a Subscription after a reconnect was a PublishRequest. What that costs is a loss window rather than guaranteed loss: the reactive repair does recover the gap when the first post-reconnect PublishResponse reveals it, and fails when the Server's retransmission queue -- whose spec minimum is only twice the Publish requests per Session, Part 4 §5.14.5.1 -- has already evicted the missed messages, which is exactly what letting Publish go first makes possible. Part 4 §5.14.1.1: "In the case of a retransmission queue overflow, the oldest sent NotificationMessage gets deleted." The client then falls through to notifyNotificationDataLost(), so the data is reported only as lost. 2. TransferResult.availableSequenceNumbers was discarded. SessionFsmFactory's transfer handling read only result.getStatusCode() (SessionFsmFactory.java:1512-1527). Part 4 §5.14.7.1 gives each successful TransferResult "the sequence numbers of the NotificationMessages that are available for retransmission" and asks that "the Client should acknowledge all Messages in this list for which it will not request retransmission" -- a should, not a shall, so this half is an improvement rather than a conformance fix. The list is nevertheless exactly the input §6.7's drain needs: it says which NotificationMessages can still be collected. The fix PublishingManager gates PublishRequests on per-Subscription reconnect recovery. sessionActivations counts Session activations and recoveredActivations records the highest one whose recovery has finished; maybeSendPublishRequests sends only while the second has caught up with the first. That test is made once the Session is in hand rather than on the way in, so a caller that was parked on getSessionAsync cannot send the instant a Session arrives either. Each Subscription is recovered on its own processing queue, paused for the duration exactly as the reactive gap repair pauses it, which is what orders the NotificationMessages the loop collects ahead of any PublishResponse for the same Subscription. lastSequenceNumber advances as they are recovered -- otherwise the next PublishResponse re-detects the gap the loop has just closed and delivers them a second time -- and each is queued for acknowledgement. Where a transfer occurred, SessionFsmFactory now hands the availableSequenceNumbers of each good TransferResult to the PublishingManager, inline, while that Session is still on its way to Active, and the loop follows the list: it starts at the oldest sequence number the Server said it holds that the client is missing, reports anything older than that as lost data, and ends where the list does rather than spending a round trip to be told what the Server has already said. Where the Session was merely re-activated there is no such list and the loop is the one §6.7 describes, incrementing until Bad_MessageNotAvailable and bounded by DEFAULT_MAX_RECOVERABLE_GAP. Requests are built against the Session that has just become Active rather than made through OpcUaClient.republishAsync, which resolves the Session again for every call: a recovery that outlives its Session must not have its remaining requests re-issued on the next one, nor park waiting for a Session that does not exist yet. A Subscription whose transfer failed is not recovered -- notifyTransferFailed resets it, which unregisters its entry, and the loop stops on an unregistered entry. Trading a loss window for a permanent stall is the risk in gating Publish at all, so every branch reaches resumePublishing: no registered Subscriptions completes at once; a processing queue that refuses the task completes at once; the loop body throwing is caught, the queue resumed and the recovery reported finished; every Republish failure, Bad_MessageNotAvailable included, ends that Subscription's loop instead of failing its future, so no per-Subscription future ever completes exceptionally; recoverSubscriptions throwing synchronously is caught and publishing resumes anyway; and recoveredActivations is a maximum, so the newest activation's recovery opens the pipeline whatever became of an older one it superseded. The tests PublishReconnectRecoveryTest drives both reconnect paths against a Server whose Publish, Republish and TransferSubscriptions responses are scripted and whose retransmission queue is modelled with a bounded capacity, so an eviction happens only when a Publish request is answered -- the causal link the ordering requirement is about. Three of its eight tests failed before this change. It asserts, on the re-activation path, that the first request to reach the Server after the Session becomes Active again is the Republish for the next expected sequence number rather than a Publish; on the transfer path, that every sequence number the TransferResult advertised is requested before the first Publish; and, with a retransmission queue that the next PublishResponse overflows, that the NotificationMessages generated while the Session was down still reach the application exactly once and in order with nothing reported lost -- the consequence the ordering exists to prevent. Its controls prove the fixture really resumes Publish on both paths, that TransferSubscriptions is reached, and that the same script against a roomier queue loses nothing even without the fix, which isolates eviction as the cause. Two further tests hold the pipeline to account when recovery cannot complete: Publish resumes after every Republish fails with a service error, and after Republish reports the Subscription is gone. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
The defect ca62339 made each of OpcUaSubscription's lifecycle transitions atomic by synchronizing the whole method body on a new lifecycleLock, so the lock was held across the blocking service call: create() (OpcUaSubscription.java:193-236, call at 204), modify() (266-317, call at 283), delete() (380-400, call at 388), setPublishingMode() (994-1022, call at 1004), and reset() (1501-1525), which makes no call of its own. reset() is not only an application entry point. PublishingManager's Bad_Timeout branch calls notifyStatusChanged (PublishingManager.java:1408-1414), which resets the Subscription (OpcUaSubscription.java:1693-1696) while running on that Subscription's deliveryQueue, and the Session FSM calls notifyTransferFailed (SessionFsmFactory.java:1546, 1569), which resets it too (OpcUaSubscription.java:1719-1720). Both therefore waited out a full request timeout - 60s by default - behind an unrelated in-flight create(), modify(), delete() or setPublishingMode(): notification delivery for that Subscription stopped for the duration, and so did the state machine that owns reconnection. On a bounded transport executor it was not a stall but a deadlock. The delivery queue is backed by that executor and so is response completion, so with a single-threaded executor - which OpcTcpClientTransportConfigBuilder.setExecutor exists to allow - the only thread that could complete the in-flight call's response was the one blocked inside reset() waiting for the lock that call held, and handleResponse had already cancelled the request timeout that would otherwise have broken the cycle. The fix lifecycleLock now guards the state transitions only, never a service call, which is what the original finding called for. A transition validates the current state under the lock, claims the right to change it by setting a private transitionInFlight flag, captures the private incarnation counter and releases the lock; it makes its service call; it takes the lock again to apply the result, and clears the claim in a finally. reset() takes the lock, does its purely local work and returns: it never waits for the Server, and never waits for a transition that is waiting for the Server. SyncState is unchanged. It is public API returned by getSyncState(), so the transitional state lives in transitionInFlight instead. getSyncState() reports the state the Subscription had when the in-flight call was made - INITIAL while it is being created, UNSYNCHRONIZED while it is being modified - which is what remains true of it until the Server answers, and is why a failed transition has no state to roll back. That is now documented on getSyncState(). incarnation is incremented by every reset() and re-compared when a response arrives. A change means the Subscription the call was made for is gone, so the result is discarded rather than applied to whatever the object holds now, and the call fails Bad_InvalidState. For create() that alone would leak: the Server has already created a Subscription, and Part 4 5.13.8 gives DeleteSubscriptions the SubscriptionId as its only handle on one, so a superseded create() now issues a best-effort deleteSubscriptionsAsync for the id it is not going to install and warns if that fails. reset()'s documented guarantee is preserved rather than revised: a reset still either discards the Subscription that existed before it, or the one an in-flight create() goes on to create; what it no longer does is wait for that call to finish first. Everything else the lock protected is kept. Transitions remain serialized against each other - a second one waits on the monitor, which releases it, instead of blocking inside a synchronized body - so a concurrent create() is still answered Bad_InvalidState and cannot create a second Subscription nothing can name. create() still installs serverState before syncState, and still registers with the client and the PublishingManager inside the critical section, so the PublishingManager entry is bound to the id create() just installed. modify()'s failure path still restores the pending Modifications diff, unless a reset() cleared them on purpose. The watchdog is still created by create() and still re-armed only after modify() has installed the revised parameters. The tests SubscriptionLifecycleContentionTest parks a create() or a setPublishingMode() inside the Server handler and asserts that work which does no I/O of its own still completes within 2s: reset() and notifyTransferFailed() on their own threads, and a scripted Bad_Timeout StatusChangeNotification reaching onStatusChanged through the delivery queue - the last of those both on an unbounded cached pool, the shape of the default executor, and on a single-threaded executor, where it also asserts the parked setPublishingMode() returns once the Server has answered rather than deadlocking. One test releases the gate and asserts the Server is left running no Subscription the client cannot name, which is the create()-superseded-by-reset() leak. Four controls fix the timing assertions in place: reset() with nothing in flight, the same Bad_Timeout script with nothing in flight on both executors, and a guard that a second create() made while the first is in flight still fails Bad_InvalidState and still leaves exactly one Subscription on the Server. modify() and delete() holding the lock across their own service call have no dedicated test: they are the same statement shape as setPublishingMode(), and a third and fourth gated-service variant would assert the same thing. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
The defect createAsync() (OpcUaSubscription.java:323-334), modifyAsync() (436-447), deleteAsync() (542-553) and setPublishingModeAsync(boolean) (1255-1266) were each their own blocking counterpart handed to FutureUtils.supplyAsyncCompose (FutureUtils.java:29-32) with the transport executor, i.e. CompletableFuture.supplyAsync(blockingCall, transportExecutor). The blocking call therefore ran on a transport executor thread and waited there for a service response - and that response is completed by the transport executor: AbstractUascClientTransport.handleResponse dispatches everything that is not a PublishResponse via config.getExecutor().execute(...) (AbstractUascClientTransport.java:149). So each of these calls occupied one thread of the pool that had to do the work it was waiting for. With a single-threaded executor - which OpcTcpClientTransportConfigBuilder.setExecutor exists to allow - the only thread that could complete the response was the one blocked waiting for it, and the wedge was permanent rather than merely slow: handleResponse cancels the request timeout (:143) before dispatching, so the wheel timer that rescues a Server which never answers cannot rescue a Server which does. On any bounded executor it was the same arithmetic: n outstanding calls occupied n threads, and once they occupied all of them none of their responses could be completed. Two concurrent createAsync() calls were enough on a two-thread pool. The fix Each transition is now implemented once, as its asynchronous form, composed from client.createSubscriptionAsync, modifySubscriptionAsync, deleteSubscriptionsAsync and setPublishingModeAsync; the blocking form is that stage awaited, so the two cannot drift apart in how they handle the lifecycle state. Nothing is dispatched onto the transport executor to wait there. A blocking form is only ever entered from an application thread: no SDK code calls create(), modify(), delete() or setPublishingMode(boolean), whose only callers outside this class are the client examples and the tests. reset() does have SDK callers on the delivery queue, on the Session FSM and in the watchdog, but it makes no service call and is unchanged. The machinery 3358c5d introduced is reused rather than duplicated: a transition claims the transition slot, makes its call with lifecycleLock released, and applies the result under the lock, comparing the incarnation it captured against the current one. What changed is how a transition waits for its turn. Waiting on the lifecycleLock monitor would have reintroduced the same defect one level down - a queued asynchronous call would still hold a transport executor thread - so transitionWaiters holds one incomplete CompletableFuture per waiting transition, and endTransition() hands the slot straight to the one that has waited longest by completing it. Nothing blocks, and no transition needs a thread of its own. Behaviour is preserved. A concurrent create() is still answered Bad_InvalidState; a create() superseded by a reset() still deletes the Subscription the Server made for it rather than leaking it, which Part 4 §5.13.8 makes the only thing that can, since the SubscriptionId is DeleteSubscriptions' only handle on one; modify()'s failure path still restores the pending Modifications diff unless a reset() cleared it on purpose; the watchdog is still created by create() and still re-armed only after the revised parameters are installed; the PublishingManager entry is still registered in the same critical section that installed the SubscriptionId it binds to. The four ...Async() methods keep their signatures and fail with the UaException the blocking form throws - await() rethrows it rather than re-wrapping it, so the StatusCode and the message are the same either way. The tests SubscriptionAsyncLifecycleTest asserts that each of the four stages completes within 5s against a Server that answers immediately on a single-threaded transport executor, and that each one had its effect: the SubscriptionId installed and SYNCHRONIZED; the revised PublishingInterval installed from an UNSYNCHRONIZED precondition; isPublishingEnabled() false; INITIAL with no Subscription left on the Server. A fifth drives two concurrent createAsync() calls on a two-thread pool. A failure there manifests as a timeout because there is nothing else to observe - the stage neither completes nor fails - so seven controls hold those assertions in place: the same four calls on an unbounded cached pool, which is the shape of the default executor; two concurrent createAsync() calls there; client.createSubscriptionAsync on the same single-threaded executor, which proves one thread is enough for the round trip and that spending it on a blocking wait is what was not; and createAsync() on a Subscription that already exists still failing with Bad_InvalidState, which pins the exceptional-completion contract the rewrite routes an error through. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Fixes from an adversarial review of this branch. Each defect is a window or failure path in machinery the branch introduced; each fix preserves the behavior the branch's tests assert. Publish suspension gate. isPublishingAllowed() compared two counters that only the onSessionActive callbacks advance, but the Session FSM completes the session future in a separate executor task submitted before the callback fan-out (SessionFsmFactory.java:668 vs :713), so a callback parked on getSessionAsync() could send PublishRequests on the new Session before recoverAndResumePublishing() counted the activation, bypassing the Part 4 §6.7 ordering this branch introduced. The gate now also records which Session the newest finished recovery ran on (RecoveredActivation), and onSessionInactive revokes it — closing both the counter window and the reactivation window, where the FSM hands back the same Session object. Watchdog during recovery. WatchdogTimer.onSessionActive re-armed the timer the moment the Session became active, but no PublishResponse — the only event that feeds it — can arrive while the gate is shut, so a recovery longer than the watchdog delay fired a spurious onWatchdogTimerElapsed on a healthy Subscription. The timer is now re-armed by resumePublishing(), when Publish traffic can actually flow. Stale work and identity. processPublishResponse re-resolved its entry from the registry by SubscriptionId, so a task queued for one registration could be applied to a different one when the Server reuses an id; it now receives the entry resolved at receipt and discards the response if that entry has been unregistered. The Bad_Timeout teardown no longer resets the live object unconditionally: it runs only if the entry is still the registered one — unregister() now reports that — and only if the object still has the incarnation the entry was registered under (resetIfIncarnation), so a stale Bad_Timeout delivered behind a slow application callback cannot tear down a replacement Subscription. Messages already received, and possibly already acknowledged, that the registration guard discards now raise onNotificationDataLost instead of disappearing at debug level. Pipeline stalls and permit leaks. A Bad_TooManyPublishRequests that returned the last outstanding request left nothing in flight to trigger a refill, halting Publish traffic until an unrelated event; the refill now runs when outstanding reaches zero. handlePublishFailure NPE'd in UaException.extract(ex) when the future completed with a non-PublishResponse (ex is null there), leaking the permit. Both rejected-task paths — processingQueue.execute() at receipt and the delivery queue submit() — now release the permit too, matching recoverSubscription's handling of the same case. Sequence accounting. A NotificationMessage further behind the expected number than any plausible duplicate — beyond max(DEFAULT_MAX_RECOVERABLE_GAP, the advertised availableSequenceNumbers) — is now treated as the Server's numbering having regressed (e.g. a restart that renumbered a restored Subscription): delivered and resynchronized to, where before it was discarded silently and indefinitely until the numbering caught back up. A gap given up on as lost data now acknowledges the advertised sequence numbers it abandons (Part 4 §5.14.7.1: acknowledge what will not be requested), so the Server's retransmission queue drains instead of holding them for the life of the Subscription. Gap repair session binding. The reactive gap repair went through client.republishAsync, which re-resolves the Session per call and parks on the one being established — the routing republish(session, ...) exists to avoid. A parked repair holds the paused processingQueue, and with it the reconnect recovery and every Subscription's Publish traffic, shut. It now binds to the Session in hand when the gap is found and treats no-Session-in-hand as lost data. Lifecycle. The parameter setters did their check-then-act on modifications and syncState without lifecycleLock, so a setter racing reset() could leave syncState UNSYNCHRONIZED with serverState null — a state every transition answers Bad_InvalidState. They now take the lock for exactly that, never across a service call. endTransition() handed the slot to the next waiter by completing its future inline, so N queued synchronously-completing transitions unwound with recursion depth N, and a StackOverflowError left the slot claimed forever; the hand-off now goes through the executor. Not changed: the pendingPublishCeiling clamp after Bad_TooManyPublishRequests still lasts until a Subscription is added or a Session activates. PublishPipelineRefillTest pins one-replacement-per-return, so probing the ceiling back up would violate the branch's stated intent. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
onPublish's direct (non-parked) path invoked responder.respondTo(request) bare, while the parked path's fulfill() converts a RuntimeException into a failed future. A scripted responder that threw on the direct path escaped into the server's async dispatch, where the response future is never completed and the client's PublishRequest hangs until its timeoutHint expires — so the same script passed or failed depending on whether the request happened to be parked first. Both paths now apply the same guard. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Regression tests for the three most serious defects 93bedb3 fixed. Each one was proved RED against the pre-fix code and GREEN at HEAD, twice over: once with both files of 93bedb3 reverted wholesale, and once with only the hunks belonging to the defect under test reverted, so that no failure is attributable to a different fix in the same commit. PublishSuspensionGateRaceTest. Pins the Publish suspension gate that af8d8da introduced to impose the Part 4 6.7 Republish-before-Publish ordering. Before 93bedb3 the gate compared two counters that only the onSessionActive callbacks advance, but SessionFsmFactory completes the Session future in one executor task and fans the activation callbacks out in another submitted after it, so a caller already parked on getSessionAsync() ran before recoverAndResumePublishing() had counted the activation and the gate answered it with the previous activation's recovery. 93bedb3 also records which Session the newest finished recovery ran on and revokes that record in onSessionInactive, because a re-activation hands back the same Session object. The window is forced, not raced: the client's transport executor has a single thread, so the task completing the Session future runs to completion -- parked continuations included -- before the activation callbacks run. The parked caller is a real one: a PublishRequest returned with Bad_UnexpectedError while the Session is down makes PublishingManager try to replace it and wait for the Session being established, and the test returns only once that failure handler has run, so the caller is provably parked. Two tests assert that the first request the Server sees after the reconnect is the Republish 6.7 requires -- one on the re-activation path, where the same Session object comes back and only the revocation closes the gate, and one on the replacement-Session path, where the ActivateSession of the new Session is held so a caller can park on the Session future the FSM will complete for it. A third test asserts the precondition on its own: a parked caller really is released before any activation callback runs. Observed pre-fix: request log [Publish, Publish, Republish:3] for both, expected Republish:3 first. SubscriptionWatchdogRecoveryStarvationTest. Pins the watchdog arming point 93bedb3 moved. WatchdogTimer.onSessionActive re-armed the timer the moment the Session became Active, but the only event that feeds the watchdog is a PublishResponse and none can arrive while the recovery gate is shut, so a recovery longer than the watchdog delay fired onWatchdogTimerElapsed on a healthy Subscription. The timer is now re-armed by resumePublishing(), when Publish traffic can actually flow. The Server holds the Republish that begins the reconnect drain until the test releases it, so the recovery lasts as long as the test says rather than as long as timing allows. With a 333ms watchdog delay (111ms publishing interval, MaxKeepAliveCount 1, multiplier 1.5) the drain is held 3s -- nine delays -- and the test asserts no expiry is reported. The control releases the held drain and asserts the expiry then arrives, which is also the assertion that resumePublishing() arms it. Observed pre-fix: the watchdog elapsed during the held drain. GapRepairSessionBindingTest. Pins the Session the reactive gap repair sends on. The repair went through client.republishAsync, which re-resolves the Session per call and parks on the one being established -- what the routing form republish(session, ...) exists to avoid. A parked repair holds its Subscription's processingQueue paused, and the Part 4 6.7 reconnect recovery is queued on that same queue, so the recovery of every Subscription is incomplete, the suspension gate never opens, and Subscriptions with nothing missing get no Publish traffic either. The Session outage is bounded by the test rather than by timing: the Server holds the reconnect's ActivateSession until the test releases it. A gap of three NotificationMessages is revealed on one of two Subscriptions, the first Republish is held, the Session is faulted, and the held Republish is then answered while no Session is available. The first test asserts the repair's remaining requests reach the Server and the whole gap plus the message that revealed it is delivered, with the reconnect still held. The second stops answering the repair's outstanding requests at the moment the reconnect completes -- so a repair already finished is unaffected and one that resumes on the new Session is held -- and asserts the other Subscription still receives its notifications. The control runs the same reconnect with no repair in flight. Observed pre-fix: only Republish 1:2 ever reached the Server, and the second Subscription received nothing after the reconnect. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Four tests pinning four of the fixes in 93bedb3 ("Fix races and stall paths in publish recovery and lifecycle"). Each was proven RED by reverting both files that commit touched -- OpcUaSubscription.java and PublishingManager.java -- to 93bedb3~1 and running the new tests against the current test tree. The wholesale revert compiled; every control test passed under it, and all five pre-existing PublishPipelineRefillTest tests stayed green under it, so each failure below is attributable to the fix its test is about. PublishPermitLeakTest, method publishRequestsResumeAfterAnAnswerThatIsNotAPublishResponse Pins the handlePublishFailure NPE. A PublishRequest that completed with a response that was not a PublishResponse reached the failure path with a null exception, and UaException.extract(null) threw before the pending-publish permit was released. The answer is fabricated by a transport-level interception, via the new TestClient.createWithTransport, because the type of a Publish answer is fixed by SubscriptionServiceSet.onPublish and the one wrong-typed message a Server can send -- a ServiceFault -- is turned into an exception by the client's channel before the SDK sees it. With maxPendingPublishRequests = 1 the leaked permit is the whole pipeline, so the test asserts a further PublishRequest reaches the Server. Pre-fix the Server saw no Publish request at all, and the run logged the defect itself: "NullPointerException: Cannot invoke Throwable.getCause() because ex is null" at UaException.extract -> PublishingManager.handlePublishFailure. The control answers the same interception with a real service failure -- the same path with an exception to take a StatusCode from -- and passes either way. PublishPermitLeakTest, method publishRequestsResumeWhenTheDeliveryQueueCannotAcceptTheNotificationMessage Pins the delivery-queue half of the rejected-task permit leak. deliverAndReleasePendingPublish releases the permit from the delivery callback, so a delivery queue whose submit() returns null -- shut down, or its max queue size exceeded -- left nothing to release it. The test shuts the Subscription's delivery queue down before the PublishResponse arrives and asserts Publish traffic continues. Pre-fix: "the client sent no further PublishRequest after a NotificationMessage its delivery queue could not accept". The control leaves the queue running. The commit's other rejected-task path, processingQueue.execute() returning false at receipt, is not pinned: the processing queue is created inside SubscriptionDetails with an unbounded max queue size, and nothing in the SDK shuts it down or exposes it, so no test can make it reject work. PublishPipelineRefillTest, method publishRequestsResumeWhenBadTooManyPublishRequestsReturnsTheLastOne Pins the Bad_TooManyPublishRequests stall. A fault that returned the last outstanding request left nothing in flight whose completion would refill the pipeline, so Publish halted until an unrelated event. Exactly one outstanding is reached with maxPendingPublishRequests = 1, which makes the target min(1 + 1, 1) = 1 and the ceiling the fault installs max(1, 0) = 1, i.e. the target the client already had. Pre-fix: "the Publish pipeline was left empty". The control faults the same one-deep pipeline with Bad_InternalError. This does not contradict TooManyPublishRequests, which pins one-replacement-per-return with two requests outstanding and stayed green throughout; the fixture grew a per-instance pipeline target so that both depths can be scripted. SubscriptionStaleDeliveryTest, method aStaleBadTimeoutDoesNotTearDownTheSubscriptionCreatedWhileItWasBeingDelivered Pins the Bad_Timeout teardown guard. deliverNotificationMessage checks that the message's Subscription still exists once, before the first callback, and then walks the NotificationData in order, so a DataChangeNotification ahead of a Bad_Timeout StatusChangeNotification puts an application callback of arbitrary duration between the check and the teardown. The test blocks in that callback, resets and re-creates the Subscription, and asserts the replacement is intact. Pre-fix the teardown was applied unconditionally: "a Bad_Timeout StatusChangeNotification received on Subscription 16 tore down Subscription 17, which was created while that message was being delivered and has never timed out". The control runs the same two-notification message with no reset in the middle of it and requires the teardown to still happen. This is the mid-message window, not the one SubscriptionIdentityTest covers, where the whole message is already stale when its delivery begins. Only the unregister()-reports-it half of that guard is pinned. The resetIfIncarnation half is reachable only when unregister() wins a race against reset()'s own removeSubscription() -- reset() bumps the incarnation before it unregisters anything -- which a test cannot drive deterministically. SubscriptionStaleDeliveryTest, method aDiscardedNotificationMessageIsReportedAsLostData Pins the reporting of notifications the registration guard discards. The client acknowledges a NotificationMessage when it is received, so a message discarded on its way to the application is data the Server may already have deleted; it disappeared at debug level. The test queues a data change behind a blocked callback, resets and re-creates the Subscription, and asserts onNotificationDataLost. Pre-fix "the application was never told". The control asserts that the baseline reports no loss. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Regression tests for four defects 93bedb3 repaired: two in the NotificationMessage sequence accounting and two in the OpcUaSubscription lifecycle. Each was proved RED against the pre-fix code and GREEN at HEAD, twice over: once with both files of 93bedb3 reverted wholesale, and once with only the hunks belonging to the defect under test reverted. In each per-hunk run the other three tests stayed GREEN, which is the isolation proof. PublishSequenceRegressionTest. Pins the boundary between a duplicate and the Server's numbering having regressed. Before 93bedb3 every NotificationMessage behind the expected sequence number was discarded as a duplicate, so a Server that restarts and renumbers a restored Subscription -- Part 4 5.14.1.1 numbers NotificationMessages per Subscription from 1 -- has every one of its messages dropped, silently, until the numbering catches back up to where it was, which for a renumbering is never. A message further behind than max(DEFAULT_MAX_RECOVERABLE_GAP, the advertised availableSequenceNumbers) is now treated as a regression: delivered, and resynchronized to. Four tests, each scripting PublishResponses whose DataChangeNotification carries its own sequence number as its value, so the delivered values say exactly which messages arrived. lastSequenceNumber is seeded reflectively, as PublishSequenceRecoveryTest does for the rollover boundary, because it only ever advances one message at a time and there is no wire path to a value hundreds ahead of where the numbering restarts; everything after the seed is the production path. One test drives a message 1000 behind and asserts it and the two contiguous messages after it are all delivered, in order, with no Republish and no lost-data report. Two more sit on either side of the boundary: 64 behind is still a duplicate and must not be delivered, 65 behind is a regression and must be. The fourth is the control with no seeding at all -- a retransmitted copy of a message received a moment ago is still discarded. Observed pre-fix: "the NotificationMessage carrying sequence 1 was never delivered", and the same for the 65-behind message. Both duplicate controls passed, so the fix did not turn every duplicate into a resync. AbandonedGapAcknowledgementTest. Pins the acknowledgement 93bedb3 added when a gap is given up on as lost data. Part 4 5.14.7.1: "The Client should acknowledge all Messages in this list for which it will not request retransmission." Nothing in an abandoned gap will ever be requested, and an unacknowledged NotificationMessage stays in the Server's retransmission queue -- re-advertised in every PublishResponse, holding its memory -- for the life of the Subscription. NotificationMessage 1 is received, then 10 arrives advertising only 2, 3 and 10: an eight-message gap inside a two-message retransmission queue, so missingSequenceNumbers abandons it without requesting any of it. The test asserts 2 and 3 are acknowledged, and that the gap really was the abandoned kind -- lost data reported, no Republish issued. This deliberately acknowledges messages that were not received, which pulls against the rule 30a3661 established that the client never acknowledges what it did not receive. The two are kept consistent by the second test, which is the companion control: a one-message gap inside the same two-message queue is recoverable, Republish(2) is attempted and fails, and sequence 2 must remain unacknowledged. The distinction is not whether the message arrived but whether the client might still ask for it. The existing acknowledgement tests -- PublishAcknowledgementTest and PublishSequenceRecoveryTest.InitialKeepAlive -- both cover recoverable gaps and are untouched. Observed pre-fix: "sequence 2 was advertised as available for retransmission and the client has decided never to request it ... expected <true> but was <false>", with the log confirming the abandoned path was taken: "Gap of 8 NotificationMessage(s) starting at sequenceNumber=2 exceeds the 3 the Server can retransmit". SubscriptionTransitionHandoffTest. Pins the transition hand-off, a defect introduced by 3358c5d / d5e515a. endTransition() handed the transition slot to the next waiter by completing its future inline, on the stack of the transition that was finishing. A transition that waits for the Server unwinds that stack, but modifyAsync() with nothing pending to send returns an already-completed stage, so N of those queued behind one another unwound with recursion depth N -- and a StackOverflowError there leaves the slot claimed with nobody left to release it, freezing the lifecycle for good. The hand-off now goes through the transport executor. A create() parked inside the Server's CreateSubscription handler holds the slot while 10,000 modifyAsync() calls queue behind it. The test asserts they all complete, that none of them reached the Server -- the premise, since synchronous completion is what recurses -- and that a real modify() made afterwards still runs, i.e. the slot was released. N comes from measurement: pre-fix, 500 drained cleanly and 1000 did not, with the hand-off stopping after 883 and 916 transitions, so the threshold on this JVM is around 900 and 10,000 is an order of magnitude past it. Observed pre-fix: "883 of 10000 queued transitions completed within 30000ms" (923 of 10000 in the per-hunk run). The StackOverflowError itself is swallowed by the CompletableFuture machinery, which is precisely why the freeze is silent. SubscriptionParameterResetRaceTest. Pins the lock the parameter setters now take. They did their check-then-act on the pending modifications and the SyncState without lifecycleLock, while reset() holds it throughout, so a setter that read the state before a reset and wrote it after left the object UNSYNCHRONIZED with no ServerState: create() answers Bad_InvalidState because the state is not INITIAL, modify() and delete() because there is no ServerState to name a Subscription with. The object is then permanently unusable. reset() is not only an application call -- PublishingManager makes it on a Bad_Timeout StatusChangeNotification and the Session FSM on a failed transfer -- so this is ordinary SDK behavior racing an ordinary application call. Not a deterministic interleaving, and it cannot be made one: the window is the setter's own read-modify-write, and a setter performs no I/O and takes no callback, so there is no seam a test could park it in. It is a bounded stress loop instead -- 40 rounds of create, race, reset, two loopback round trips each -- which provokes the window efficiently because the window is nearly the whole of the setter's body, so a thread calling it in a loop is inside it a large fraction of the time. The resulting state is sticky too: every later setter call re-marks it. Each round stops and joins the setter threads before asserting, so nothing is read while a background thread races to update it, and the Server-side Subscription the reset orphaned is deleted so the loop leaks nothing. Observed pre-fix: round 2 of 40 -- round 1 in the per-hunk run -- left the Subscription "UNSYNCHRONIZED with SubscriptionId Optional.empty". The whole test takes about 0.25s and passed five consecutive runs at HEAD. Revert method: both, for all four tests. The wholesale revert of OpcUaSubscription.java and PublishingManager.java to 93bedb3~1 compiled against the current test tree and failed exactly the five expected tests of the eight in these four classes, and nothing else. The per-defect reverts applied individual hunks of 93bedb3 in reverse: PublishingManager hunk 12 of 25 for the sequence regression, hunks 14 and 15 for the abandoned-gap acknowledgement, OpcUaSubscription hunk 3 of 11 for endTransition, and hunks 4 through 8 for the five parameter setters. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Part 4 §5.14.5.1: once a Server answers Bad_TooManyPublishRequests a Client
"shall not issue another Publish request before one of its outstanding Publish
requests is returned", and PublishingManager also learns a ceiling from the
fault — the number of requests the Server was in fact holding — so that the next
returning request does not restore the very outstanding count that drew it.
That clamp used to last until a Subscription was added or a Session activated,
and neither need ever happen again in the life of a long-lived client. A merely
transient condition on the Server — a queue briefly full, a cap raised again a
second later — therefore cost such a client a permanently shallower Publish
pipeline, i.e. throughput, for no reason. The same clause requires a Server to
accept at least subscriptionCount + 1 queued Publish requests, which is exactly
what the client aims for, so a ceiling only ever describes a Server that is
non-conformant or momentarily overloaded; in the second case the client ought to
find out that the moment has passed.
The ceiling now recovers, driven by the Server's own answers rather than by a
clock: releasePendingPublish(), the one place that knows a PublishRequest went
out and came back without being refused, counts successful Publish round trips
against it. Counting successes rather than elapsed time means a Subscription
that is publishing pays for its probes, a silent one does not probe at all, and
the policy is deterministically testable without a clock.
* Recovery is incremental: a probe raises the ceiling by one, never straight
back to min(subscriptionCount + 1, maxPendingPublishRequests), because
closing the whole deficit at once would send the Server exactly the burst
that drew the fault. The natural target is never exceeded.
* A probe costs a cooldown of successful round trips —
PROBE_COOLDOWN_BASE_SUCCESSES = 8 to begin with. Eight is comfortably clear
of the one-returning-request window §5.14.5.1's rule is measured in, so the
immediate response to the fault is untouched, while still being few enough
NotificationMessages that a Subscription publishing once a second reaches
its first probe in under ten seconds. The cooldown is also what keeps a
single probe in flight: the one extra request a raise adds is answered long
before another whole cooldown of requests has been.
* A probe the Server refuses re-clamps the ceiling and multiplies the next
cooldown by PROBE_COOLDOWN_GROWTH = 2, up to PROBE_COOLDOWN_MAX_SUCCESSES =
512. The cap is there because the alternative to a bounded backoff is one
that gives up; at 512 a Subscription publishing once a second probes about
every eight minutes in the worst case, which costs a Server that will never
accept more essentially nothing.
The anti-hammering guarantee this buys, stated exactly: a Server that always
answers Bad_TooManyPublishRequests sees a number of probes that grows only
logarithmically in the number of responses it delivers — they land at roughly 8,
24, 56, 120, 248, ... successful round trips, so doubling the traffic adds one
probe rather than doubling the probes, and no interval between two probes is
ever shorter than the interval before it.
The state is one immutable value (PendingPublishCeiling) behind a single
AtomicReference rather than four fields, so a refusal cannot interleave with a
raise. Its "probing" flag is deliberately not cleared by the first successful
round trip after a raise: a refused PublishRequest and a delivered
NotificationMessage are handled on paths of different lengths, and in practice
the delivery wins, so a flag the delivery could clear would leave the refusal
behind it uncharged — which is precisely what a Server that always refuses
produces, and would have it probed at a constant rate forever. The flag is
instead cleared by the refusal charged to it, or by the raised ceiling standing
for a whole cooldown, which also stops unrelated transient conditions far apart
in time from ratcheting the cooldown up to its cap.
Tests
-----
PublishCeilingRecoveryTest drives a ScriptableSubscriptionServiceSet subclass
that models a Server which will not queue more than a settable number of
PublishRequests, answering one parked request at a time, so the number of
requests the Server is left holding after each answer is the client's ceiling,
observed rather than inferred. Each fixture states its natural target rather
than inheriting one: maxPendingPublishRequests is configured equal to it, so one
Subscription targets min(1 + 1, 2) = 2 and three target min(3 + 1, 4) = 4.
* Recovery: one refusal while filling a two-deep pipeline clamps the ceiling
to one; once the cap is lifted, sustained keep-alive traffic brings the
outstanding count back to the natural target. Incrementality is asserted
separately, against a four-deep pipeline clamped to two by two refusals,
whose recovery must be observed as three outstanding and then four rather
than four at once.
* Anti-hammering, the important one: against a Server that answers
Bad_TooManyPublishRequests to anything above a fixed cap, 256 delivered
NotificationMessages must draw at least two and at most ten refused probes,
with the intervals between them never shrinking. Measured: five, at 8, 25,
58, 123 and 252 round trips.
* No regression of the immediate behavior: over the first four returning
requests after the clamp each still buys exactly one replacement, and the
ceiling does not move. PublishPipelineRefillTest, which pins
one-replacement-per-return for the first of them, is unchanged and green.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
PublishingManager ran the Part 4 6.7 Republish drain on every Session activation, including the client's first. recoverSubscriptions() snapshots subscriptionDetails when the onSessionActive callbacks run, and connect() returns when the Session future completes -- which happens in a task submitted before the callback fan-out (SessionFsmFactory.java:668 vs :713). A Subscription created immediately after connect() can therefore register itself inside that window and be included in a recovery that has nothing to recover: the Session and the Subscription were both created moments earlier, so the Server cannot be holding a NotificationMessage the client has not collected, and every Republish the drain sends can only be answered Bad_MessageNotAvailable. The cost is not only the wasted round trips. The drain holds the publish gate shut until it finishes, see isPublishingAllowed(), so it also delays the first PublishRequest of every Subscription caught in the window -- and a drain whose Republish is slow to be answered holds all Publish traffic with it. Recovery is now skipped unless a Session has previously become inactive, or a Subscription carries the availableSequenceNumbers a transfer named. Both are evidence that the Server may hold something the client has not seen; the first activation of a Session the client just created is not. The narrower test "lastSequenceNumber == 0" was deliberately not used: a Subscription that lived through a disconnect having received nothing still needs the drain. This was reaching CI as unrelated flakes. With the whole integration-tests module in one JVM fork on two CPUs the window opens often enough to produce a spurious Republish in PublishSequenceRecoveryTest and PublishResponseOrderingTest, a reset ceiling in PublishCeilingRecoveryTest (onSessionActive clears it by design), and a Publish that is never sent in SubscriptionWatchdogRecoveryStarvationTest, whose held Republish responder kept the gate shut so the Bad_SessionIdInvalid fault that test scripts was never delivered and its reconnect never happened. That last one is the failure seen on CI. Three runs of that configuration are green with this change; two of two reproduced a failure without it. PublishFirstActivationRecoveryTest asserts that no Republish is requested on the first activation and, as the control, that one is requested after the Session is re-activated. It is a guard rather than a reproduction: entering the window needs the calling thread to beat the fan-out task, which it loses on an idle machine, so the class also passes against the unfixed code. That is recorded in its Javadoc. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
…rdening # Conflicts: # opc-ua-sdk/integration-tests/src/test/java/org/eclipse/milo/opcua/sdk/test/TestServer.java
Two tests asserted on state reached before the event they were waiting for, and failed under the contention of the whole integration-tests module running in one fork -- and now, since forkCount=2 landed on main, of two forks sharing a machine. Both are test defects; neither changes production behaviour. SubscriptionStaleDeliveryTest waited for the Subscription to reach INITIAL and then asserted the Bad_Timeout had been reported. notifyStatusChanged() resets the Subscription before it tells the application, so INITIAL is reached first and the wait can return with statusChanges() still empty -- observed failing with the whole assertion taking 22ms. It now waits for the report, which is the later of the two, and checks INITIAL afterwards without a wait, since the reset has necessarily already happened by then. PublishCeilingRecoveryTest asserted that every interval between refused probes was at least as long as the one before it. The interval is measured in round trips a refusal was recorded against, and a refusal and the delivery it is attributed to are separately timed: the fault path takes two executor hops the response does not, so which round trip a refusal lands against jitters by one either way. An observed [8, 16, 1, 32, 64, 128] is that jitter, not a cooldown that failed to grow. The assertion now pins the direction -- the window ends with a longer wait than it began with -- and leaves the guarantee itself to the bound on the probe count above it, which is unaffected by attribution. Four runs of the Subscription tests in one fork are green, where the previous assertions failed in one of two, and the full build passes. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Merged
transferSubscriptions() built the id array by flatMapping only the Subscriptions that currently hold a SubscriptionId, but the failure fan-out indexed the unfiltered snapshot by TransferResult index. A Subscription reset() concurrently -- after the snapshot was taken but before the ids were collected -- shifted the pairing by one, so a bad result could be applied to the wrong Subscription and notifyTransferFailed() would reset one that had transferred successfully. A Server returning more results than requested ids threw IndexOutOfBoundsException instead, skipping every remaining onTransferFailed notification. The ids sent and the Subscriptions they came from are now captured together in one pass, and both result loops index that aligned list, bounds-guarded the way the good-result loop already was. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
resetWatchdogTimer() armed unconditionally, so a create or modify made while the reconnect recovery held the publish gate shut -- and the finish of a stale recovery that a newer activation had superseded -- armed a timer that no PublishResponse, the only event that feeds it, could possibly reach. A recovery outlasting the watchdog delay then fired onWatchdogTimerElapsed on a Subscription that was recovering normally, and a typical application handler tears such a Subscription down mid-recovery. resetWatchdogTimer() now asks the PublishingManager whether Publish traffic is suspended and defers arming while it is; the recovery that ends the suspension re-arms every registered Subscription's timer, as it already did. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
The Bad_SessionClosed/Bad_SessionIdInvalid branch of handlePublishFailure() paused every Subscription's watchdog with no staleness check, unlike the Bad_NoSubscription branch beside it. A straggling failure from a replaced Session could therefore run after the new Session's recovery had re-armed the watchdogs and disarm them all; if the new Server went quiet before the next PublishResponse re-armed them, the stall the watchdog exists to report went unreported. Each PublishRequest now records the activation count in effect when it was sent, and the pause only happens while that count still describes the client. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
addSubscription() registered with a plain put(), so a Server reusing a SubscriptionId while an entry was still registered under it displaced that entry without marking it unregistered. unregister() matches by (key, value) and could never remove it again, leaving registered=true forever: work still queued for the displaced entry was applied as if its Subscription existed, indefinitely. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
processPublishResponse() queued the acknowledgement before the duplicate check and without the contains() guard that restoreAcknowledgements() and acknowledgeAbandonedSequenceNumbers() both apply. A duplicate copy arriving before the first acknowledgement was drained queued the same (subscriptionId, sequenceNumber) twice; the Server deletes the message on the first and answers the second Bad_SequenceNumberUnknown, producing recurring spurious refused- acknowledgement warnings. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
kevinherron
marked this pull request as ready for review
July 30, 2026 01:50
Keep lifecycle cleanup independent of caller cancellation, close watchdog and monitored-item registration races, and provide ordered transfer cleanup that cannot be delayed by an overridable notification.
Bind transfer inputs and Publish pipeline learning to the Session that produced them, size blind recovery from the deepest permitted pipeline, and finish failed-transfer cleanup before reconnect recovery can observe subscriptions.
Wait for reset to complete before releasing the gated create response so the test always exercises superseded-create cleanup.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Corrects client-side subscription publishing, recovery, and lifecycle handling under lost NotificationMessages, reordered Publish responses, UInt32 sequence rollover, reconnect, subscription reset and recreation, and failed service calls. Affects
PublishingManager,OpcUaSubscription, and the transfer-result handling inSessionFsmFactory.Sequence-number accounting and response ordering
Sequence arithmetic is implemented in a package-private
SequenceNumberstype using the rule in Part 4 §5.14.1.1 — sequence numbers roll over to 1 and never use 0 — rather than the generic wrap-to-0 Counter rule of §7.8, which is off by one at each rollover. Previously the arithmetic was plainlongaddition and comparison with no rollover handling, so a message lost at the boundary was never detected.A keep-alive sets
lastSequenceNumberto the modular predecessor of the advertised number, because §5.14.1.1 defines a keep-alive as carrying the sequence number of the next NotificationMessage to be sent. Previously an initial keep-alive advertising sequence 1 recorded 1 as received, masking the loss of NotificationMessage 1; and a gap ending in a keep-alive advanced only to the first missing number, so each subsequent keep-alive re-requested the remainder of the gap and reported data loss again.Publish responses are processed in the order the transport delivers them.
AbstractUascClientTransportserializesPublishResponsecompletion through a dedicated queue; the manager previously re-dispatched that callback onto the multi-threaded transport executor, so responses could be processed out of order, regressing sequence state and producing duplicate delivery and spuriousonNotificationDataLost. Processing is now enqueued from the ordered completion callback and is monotonic, so a late or duplicate response cannot move state backwards.Gap recovery is bounded by what the server advertises in
availableSequenceNumbers, falling back to a fixed maximum. A gap larger than that is reported as lost data and resynchronized to, rather than iterated — necessary because modular comparison, unlike the previous plain>, will compute a large forward distance for a stale or wrapped sequence number.A NotificationMessage further behind the expected number than any plausible duplicate — beyond
max(DEFAULT_MAX_RECOVERABLE_GAP, advertised availableSequenceNumbers)— is treated as the server having renumbered, for example after a restart that restored a subscription. It is delivered and resynchronized to, where previously it was discarded and the client remained stalled until the server's numbering caught back up.Republish recovery
Recovery is asynchronous and never blocks the executor that completes its own responses. Previously it called the blocking
client.republish()from the response-processing queue while the Republish response could only be completed by the same executor. On a single-thread executor this deadlocks permanently:handleResponsecancels the request timeout before queueing the completion, so once the server answers, nothing remains to break the stall.Response processing uses a per-subscription queue rather than one client-wide queue, so a gap in one subscription does not stall response processing for the others.
Recovery is bound to the session in hand when the gap is found. Going through
client.republishAsyncre-resolves the session per call and parks on one being established; a parked repair holds the paused processing queue, and with it reconnect recovery and Publish traffic for every subscription.Reconnect
After a session is re-activated the client performs the Part 4 §6.7 Republish drain for each subscription before resuming normal Publish handling. The drain runs only where the server may hold something the client has not collected: a session has previously become inactive, or a subscription carries the
availableSequenceNumbersa transfer named. It does not run on the first activation of a session the client just created, where a subscription registered between the session future completing and theonSessionActivefan-out would otherwise be drained for messages that cannot exist — spending a round trip per subscription and holding the publish gate shut for their duration. The SecureChannel, ActivateSession, and TransferSubscriptions parts of §6.7 were already handled; the drain was not, which allowed the server's retransmission queue to evict missed messages before the client requested them.SessionFsmFactorypasses each goodTransferResult'savailableSequenceNumbersto the drain, so after a transfer recovery follows the list the server holds rather than incrementing untilBad_MessageNotAvailable. §5.14.7.1 states the client should acknowledge messages in that list it will not request, so this is a conformance improvement against a "should", not a "shall".The SubscriptionIds sent in
TransferSubscriptionsand the Subscription objects they came from are captured together, so a concurrentreset()cannot shift result indexes and apply a failed result to a Subscription that transferred successfully. Transfer recovery input is owned by the exactUaSessionthat produced it and is consumed only by a matching recovery. A recovery task queued for an older Session cannot take the replacement Session's sequence-number list, and a late Republish response is discarded once its activation has been superseded.The suspension gate records which session the most recent completed recovery ran on, and
onSessionInactiverevokes it. A gate keyed only on activation counts can be bypassed, because the Session FSM completes the session future in an executor task submitted before the activity-listener fan-out, so a caller already parked ongetSessionAsync()can send a Publish before recovery is counted. Recording the session also covers reactivation, where the FSM hands back the sameSessionobject.Every failure path in recovery reaches
resumePublishing, including an empty registry, a rejected queue task, a throwing drain body, and any Republish failure. Drains are sent against the just-activated session withsendRequestAsyncso a drain cannot park on a future session and leave a queue paused.Where no transfer result advertises the retransmission queue, the new reconnect drain is bounded by the deepest Publish pipeline Milo has permitted, plus the terminating request that receives
Bad_MessageNotAvailable. The base branch resumed normal Publish handling without performing this drain; the bound covers every response Milo could have left outstanding while remaining finite. The pipeline high-water mark is recorded when the target is chosen, before asynchronous request failures, Session-inactive callbacks, or Subscription removal can make the old depth appear smaller.A failed
TransferResult, or an expected service-level transfer failure, resets and unregisters the affected Subscription before the transfer future can move the Session FSM through initialization and into Active. The cleanup path is final and non-overridable; the publicnotifyTransferFailed()callback is dispatched afterward, so a blocking or throwing application override cannot delay reconnect or let recovery observe a Subscription that was not transferred.Acknowledgements
Acknowledgements are built from the sequence numbers the client actually received or successfully recovered. Previously the acknowledgement set was replaced with the server's entire
availableSequenceNumbers, which §5.14.5.2 describes as diagnostic information about the retransmission queue, not evidence of receipt — so a failed Republish could be followed by an acknowledgement telling the server to delete a message the client never received and could still have recovered.A duplicate copy of a NotificationMessage queues at most one acknowledgement for its
(SubscriptionId, sequenceNumber). Without the duplicate guard, a second acknowledgement for a message the first one already deleted produced recurring spuriousBad_SequenceNumberUnknownwarnings.Acknowledgements drained into a PublishRequest are restored if that request fails, so they are not lost when the request carrying them does not reach the server.
A gap abandoned as lost data acknowledges the advertised sequence numbers it gives up on, per §5.14.7.1, so the server's retransmission queue drains rather than holding them for the life of the subscription.
PublishResponse.resultsis inspected and acknowledgement failures are logged; it was previously ignored, soBad_SequenceNumberUnknownandBad_SubscriptionIdInvalidwere invisible.Subscription identity and lifecycle
PublishingManagerentries carry an immutableSubscriptionIdand a registration flag. Entries previously held the mutableOpcUaSubscriptionand asked it for its current id, so aBad_Timeoutstatus notification queued before a reset would unregister and reset a subscription created after it — leaving a live server-side subscription with no client registration and no way to delete it, sincedelete()requiresserverState. Response processing now carries the entry resolved at receipt rather than re-resolving by id, which also covers a server reusing a SubscriptionId, and theBad_Timeoutteardown runs only if the entry is still registered and the object still holds the incarnation the entry was registered under. Messages discarded by that guard raiseonNotificationDataLost.If registering a new Subscription displaces an older entry under a reused SubscriptionId, the displaced entry is marked unregistered immediately. Work already queued for it is then discarded instead of being applied indefinitely as if the old Subscription still existed.
lifecycleLockserializescreate,modify,delete,setPublishingMode, andreset, and is held only for state transitions — never across a service call. Each transition validates and claims under the lock, performs the service call with it released, then applies the result under it again, comparing an incarnation counter so a superseded result is discarded. Holding the lock across the call deadlocks:reset()takes the same lock and can be reached fromnotifyStatusChanged(Bad_Timeout)on the delivery queue, a transport-executor thread, so a blockedcreate()stalls the delivery queue for a full request timeout on any executor, and on a bounded executor can exhaust the pool that would complete the response releasing the lock. The parameter setters take the lock for their check-then-act onmodificationsandsyncState; without it a setter racingreset()could leavesyncStateUNSYNCHRONIZEDwithserverStatenull, a combination every transition answersBad_InvalidState.Before this PR, transfer-failure reset and notification were dispatched together without making
TransferSubscriptionscompletion wait for the reset, so the Session FSM could reach Active while a failed Subscription was still registered. A finalhandleTransferFailure()now performs reset and unregistration synchronously before the FSM proceeds, then dispatches the overridable notification separately.If a
reset()supersedes an in-flightcreate(), the subscription the server already created is deleted rather than abandoned.createAsync,modifyAsync,deleteAsync, andsetPublishingModeAsyncare composed fromcreateSubscriptionAsync,modifySubscriptionAsync,deleteSubscriptionsAsync, andsetPublishingModeAsync; each blocking method awaits the corresponding stage, so the two forms cannot diverge in their state handling. Previously all four wrapped their blocking counterparts insupplyAsyncComposeon the transport executor and self-deadlocked on a single-thread executor. Queued transitions wait on handed-overCompletableFutureslots rather than by blocking a thread, and the hand-off is dispatched through the executor rather than completed inline, which would otherwise unwind with recursion depth equal to the number of queued synchronously-completing transitions.The stage returned to a lifecycle caller is now a dependent view of the internal transition stage. Cancelling it, or completing it early with
orTimeout(), does not cancel the internal completion that owns and eventually releases the transition slot; later lifecycle operations therefore cannot be wedged by caller-side cancellation.create()publishesserverStatebeforesyncState. Both classes document their threading contract.MonitoredItem deletion
deleteMonitoredItems()removes items from the pending-deletion set only when the delete produced an operation-level result — Good, orBad_MonitoredItemIdInvalid, the only other operation result §5.13.6.4 Table 77 defines — and in both cases the item is gone server-side. Previously the set was cleared before the service call, so a service fault, a nullserverState, or unresolvable MonitoredItemIds orphaned the item: it could not be retried, could not be re-added (addMonitoredItemignores an item holding a handle it does not own), andisMonitoredItemsSynchronized()reportedtruewhile the server kept publishing for it.MonitoredItem add/remove bookkeeping and application of delete results are ordered under one lock. If an item is added back while its server deletion is in flight, the response clears the obsolete server state but restores the ClientHandle under which the item was re-added, leaving it mapped and ready to be created exactly once.
reset()resets and clears the pending-deletion set. Items left in it retained their MonitoredItemIds andSYNCHRONIZEDstate, and a laterdeleteMonitoredItems()sent those ids against the new subscription. Milo's server allocates MonitoredItemIds per subscription starting at 1, so a stale id generally matches an unrelated item in the replacement subscription and the delete succeeds against the wrong item.Publish pipeline accounting
The
getTimeoutHint()overflow guard clamps the computed timeout. It previously assigned a different variable, so an out-of-range value reacheduint()and threwNumberFormatExceptionafter the pending permit was taken and before any completion handler was registered, leaking the permit; repeated attempts pinned the counter and stopped Publish traffic. Request construction now releases its permit on every synchronous failure path, as do the rejected-task paths at receipt and on the delivery queue.handlePublishFailureno longer callsUaException.extractwith a null throwable, which threw and leaked the permit when a future completed with a non-PublishResponsevalue.Refill after
Bad_NoSubscriptionis suppressed only while the subscription set is still empty, so deleting and recreating the last subscription does not leave the pipeline with no trigger to refill it. Refill also runs when the outstanding count reaches zero, so aBad_TooManyPublishRequestsreturning the last outstanding request does not halt Publish traffic until an unrelated event.The ceiling applied after
Bad_TooManyPublishRequestsrecovers rather than persisting for the life of the client. It rises one step afterPROBE_COOLDOWN_BASE_SUCCESSESsuccessful round trips; a probe that draws another refusal clamps the ceiling again and doubles the next cooldown, up toPROBE_COOLDOWN_MAX_SUCCESSES. One probe is in flight at a time and the ceiling never exceedsmin(subscriptionCount + 1, maxPendingPublishRequests). A server that consistently refuses therefore sees a probe count growing logarithmically in successful responses. §5.14.5.1's requirement not to issue another Publish request before an outstanding one returns is unaffected: recovery is separate from the one-replacement-per-return handling of the refusal itself.The learned ceiling is tagged with the Session activation that produced it. A delayed
Bad_TooManyPublishRequestsfrom an old Session cannot clamp the replacement Session, and delayed successful responses from the old Session cannot pay down the replacement's cooldown or trigger an early probe.Notification delivery
Listener invocations are individually isolated, so one throwing listener does not suppress the others. A throwing
SubscriptionListenerpreviously prevented every monitored-item listener from seeing that notification, a throwing item listener prevented later items from being notified, and the throw escaped theNotificationDataloop, discarding the remaining elements of the same NotificationMessage. The lists passed to the subscription listener are immutable; the same mutable lists were previously reused to drive the per-item fan-out, so listener mutation could suppress it or mispair items with values.onKeepAliveReceivedandonStatusChangedare invoked inline, as data and event callbacks already were. They were re-enqueued onto the delivery queue they were already running on, which placed them behind an already-queued later notification and released Publish backpressure before they ran.onNotificationDataLostandonTransferFailedremain enqueued, as they are called from off the delivery queue.Part 4 does not specify client-side callback ordering; this is an SDK delivery-contract change.
Watchdog
modify()re-arms the watchdog after installing the revised publishing interval and counts. Re-arming first uses the pre-modifyServerState, so a longer revised interval leaves the timer armed for the old, shorter delay.A session fault suspends the watchdog rather than destroying it.
cancelWatchdogTimer()deregisters theSessionActivityListenerand clears the timer, and onlycreate()constructs one, so aBad_SessionClosedorBad_SessionIdInvalidfault left the watchdog permanently dead on a subscription that then reconnected and transferred successfully. Suspension cancels the pending expiry and leaves the listener registered; the destructive path remains for subscription teardown.The pause is activation-scoped: a straggling Session failure cannot run after replacement-session recovery and disarm the watchdogs that recovery just re-armed.
The timer is re-armed by
resumePublishing()rather than byonSessionActive, because aPublishResponseis the only event that feeds it and none can arrive while reconnect recovery holds Publish suspended — so a recovery longer than the watchdog delay fired a spuriousonWatchdogTimerElapsedon a healthy subscription.resetWatchdogTimer()also refuses to arm while publishing is suspended. Creation registers the Subscription withPublishingManagerbefore its final watchdog check, so either the recovery's registry sweep sees the new Subscription or the final check sees publishing resumed; the watchdog cannot fall between those two paths and remain unarmed.Timer transitions are serialized behind a lock with a terminal cancelled flag and an epoch validated inside the expiry callback, so an expiry already running when
cancel(false)arrives is recognized as stale, and ascheduleNext()that loses a race cancels its own future rather than leaking it.Subscription parameters
OpcUaSubscription(OpcUaClient, double)derives MaxKeepAliveCount and LifetimeCount from the given publishing interval. Field initializers run before the constructor body, so both were derived from the 1000 ms default:new OpcUaSubscription(client, 100.0)requested MaxKeepAliveCount 10 and LifetimeCount 50 instead of 100 and 500, making the keep-alive cadence and the subscription lifetime wrong byrequestedInterval / 1000and carrying the same error into the watchdog delay.A failed
modify()restores the pendingModificationsrather than clearing them before the service call, where a retry found them null. Values requested while the failed call was in flight take precedence over the restored ones.Behavior changes
Republishis issued for sequence 1 afterkeep-alive(1)followed bydata(2). Commit33438b732(SetlastSequenceNumberon the first PublishResponse #1401) suppressed this. Per §5.14.1.1 a keep-alive advertising 1 means NotificationMessage 1 has not been sent, so when data 2 arrives, 1 is missing. Milo's server sendskeep-alive(1)thendata(1)and is unaffected. Against a server that consumes a sequence number for keep-alives the cost is one failed Republish and oneonNotificationDataLostper subscription.create()fails withBad_InvalidStaterather than creating and leaking a second server-side subscription.isMonitoredItemsSynchronized()reportsfalseafter a failed MonitoredItem delete, since the deletion stays queued.getSyncState()returns the pre-call value while a transition is in flight —INITIALwhile creating,UNSYNCHRONIZEDwhile modifying.SyncStateremains a three-value public enum; this is documented on the method.Testing
mvn clean verifypasses, with spotless and checkstyle clean.New tests cover: the initial keep-alive masking a lost NotificationMessage; a gap ending in repeated keep-alives; rollover with and without a missing boundary message; reordered Publish completion via an injected reordering executor; Republish and all four async lifecycle methods on a single-thread transport executor; timeout-hint overflow and pending-permit recovery; delete-and-recreate of the last subscription with delayed
Bad_NoSubscriptioncallbacks;Bad_TooManyPublishRequestsagainst a server cap below the attempted window, including ceiling recovery and the probe bound; reset and recreate with stale delivery andBad_Timeouttasks still queued; reconnect asserting no Publish precedes the Republish drain, including with a caller parked on the session future and across both reactivation and replacement-session paths; the watchdog during a held drain; modify failure followed by retry; watchdog rescheduling after revised parameters; listener isolation and callback ordering; MonitoredItem deletion failure and stale-id reuse across a reset; abandoned-gap acknowledgement; and queued lifecycle transition hand-off.Additional regression tests cover caller-cancelled lifecycle futures; watchdog registration racing reconnect resume; a MonitoredItem re-added during an in-flight delete; stale recovery versus replacement-Session transfer data; the configured Publish pipeline depth and terminating probe; stale activation failures and successes versus the replacement Session's Publish ceiling; and transfer-failure cleanup ordering with blocking and throwing overrides.
SequenceNumbershas server-free unit tests for successor, predecessor, forward distance, and the wrap-to-1 boundary.Two fixed sub-defects have no test: the
serverState-before-syncStatepublication order, where the two statements are adjacent with no interposable call and every reader null-checksserverState; and the unresolvable-MonitoredItemIds branch of the deletion path, reachable only via areset()racing an in-flightdeleteMonitoredItems().The far-behind-sequence and rollover tests seed
lastSequenceNumberreflectively, because it advances one message at a time and there is no wire path to a value far ahead of where a renumbering restarts. Everything after the seed is driven by real responses over a real connection.A test seam supports the above:
ScriptableSubscriptionServiceSetscripts the exact Publish and Republish responses the client observes, parks pipelined Publish requests, and captures the acknowledgements the client sends;DelegatingSubscriptionServiceSetandDelegatingMonitoredItemServiceSetallow individual service handlers to be gated or failed; andTestClientexposes the transport config builder for injecting executors.🤖 Originally generated with Claude Code; updated after subsequent code review.