[#112] Deliver CancelOpRequest arriving before the operation registers

[vharseko]

Part 1 of #112: the CancelOpRequest registration race.

Problem

Operation execution is submitted to the shared pool and the LocalRequest registered itself in its constructor on that pool thread. A CancelOpRequest for the same message id processed before that registration removed nothing from localRequests and was silently dropped: the operation ran to completion (wasted work for search/sync), and for subscription operations — where the cancel is the unsubscribe mechanism — the subscription kept streaming events until the connection died.

Two related defects surfaced during the fix:

• LocalRequest published this from its constructor, so a concurrent cancel could observe a partially constructed processor (subclass fields still null) and NPE in tryCancel().
• The subscription/batch processors assign their Subscription only inside executeOperation, so even a correctly delivered early cancel hit subscription.close() on null.

Fix

• RemoteConnectionGroup: a cancel that finds no registered request is parked (request id → expiry, TTL 60 s) and applied when the operation registers. Park-then-lookup vs publish-then-check write the two maps in opposite order, so whichever side runs second sees the other — the cancel is never lost regardless of processing order. Request ids are never reused within a group, so a parked cancel can only match the operation it was sent for.
• LocalRequest: registration moved out of the constructor into an explicit register() that returns false when a parked cancel already consumed the request — the operation is then never started. cancel() is single-shot, making a duplicate delivery during the race harmless.
• OpenICFServerAdapter.processOperationRequest: all 16 dispatch branches now only build the processor; registration and execution happen in one place via processor.register() && processor.execute(...).
• ConnectorEventSubscriptionApiOpImpl, SyncEventSubscriptionApiOpImpl, BatchApiOpImpl: the Subscription hand-over happens under a lock with a cancelled flag — a cancel racing with subscribe()/executeBatch() closes the subscription as soon as it exists instead of throwing NPE.

Tests

New LocalRequestRegistrationTest covers: no registration from the constructor, register-then-cancel, cancel-before-registration is parked and applied, parked cancel does not affect other ids, cancel delivered exactly once, a deterministic replay of the interleaving where receiveRequestCancel delivers the cancel directly between the registration's putIfAbsent and its pending-cancel check, and a 1000-round two-thread stress test racing register() against receiveRequestCancel(id).

Verified: connector-framework-rpc 13/13, connector-framework-server 25/25, connector-server-jetty 30/30.

Compatibility note

LocalRequest subclasses are no longer registered by the constructor — call register() after construction and execute the operation only when it returns true. Out-of-tree subclasses relying on constructor registration compile unchanged but will not receive responses or cancel messages until updated. RemoteConnectionGroup.receiveRequest now returns null (and leaves the request unregistered) when a pending cancel already consumed the request.

[maximthomas]

Two things worth addressing before merge.

1. register() can still return true for an already-cancelled request

RemoteConnectionGroup — this interleaving is reachable:

A: localRequests.putIfAbsent(id, req)        // registers
B: pendingCancels.put(id, expiry)
B: localRequests.remove(id)      -> req      // non-null
B: pendingCancels.remove(id)                 // B consumes the park
B: req.cancel()                              // cancel delivered, OK
A: pendingCancels.remove(id)     -> null     // A sees nothing
A: return req  ->  register() == true  ->  execute() runs

The cancel is still delivered, so the headline guarantee holds and this is not a regression against master. But the invariant stated in the PR description and in the register() javadoc — "returns false when a cancel for this request id was already received" — does not hold here, and for a subscription op the connector actually subscribes before attachSubscription tears it back down.

The AtomicBoolean cancelled you added already carries the answer; make the decision depend on the request's own state instead of on winning the map removal:

// LocalRequest
public final boolean isCancelled() {
    return cancelled.get();
}

// RemoteConnectionGroup.receiveRequest
if (null != pendingCancels.remove(localRequest.getRequestId())) {
    localRequest.cancel();
}
return localRequest.isCancelled() ? null : localRequest;

This also lets LocalRequestRegistrationTest assert the invariant directly rather than inferring it.

2. No concurrent test for a concurrency fix

The five new tests cover the state machine well, but all five are single-threaded — none exercises the race the PR exists to fix, and none would have caught the window above. A stress test is cheap here:

// N iterations; per round, two threads race register() against receiveRequestCancel(id).
// Assert after each round: request.isCancelled() is true, and tryCancel() ran exactly once.

Minor

• Parked cancels linger up to 60 s when a cancel loses the race with completion. tryCancelRemote only fires from PromiseImpl.tryCancel while the promise is pending, so this is low-volume and self-purging — but the park is unconditional, so every cancel that arrives after the server already produced its result holds an entry for the full TTL. A line in the javadoc acknowledging that would help the next reader.
• receiveRequest returning null is a semantic change to published API. All in-tree LocalRequest construction sites are converted, but an out-of-tree subclass relying on constructor registration would compile unchanged and silently never register (no cancel delivery, no receiveRequestUpdate routing). Worth a release note.

[vharseko]

Thanks for the careful review — all four points are addressed in 73542c9.

1. register() returning true for an already-cancelled request — confirmed reachable exactly as traced. Took your fix as proposed: LocalRequest.isCancelled() (final, backed by the existing AtomicBoolean) and the verdict in receiveRequest now comes from the request's own cancelled state instead of from winning the park removal. One addition on top: the null-return path also does localRequests.remove(id, request), so a request reported dead can never be left registered — in your interleaving the request was also unregistered by the cancelling side while execute() still ran, so this makes the invariant "register() returns false ⟹ cancelled, not registered, never executed" airtight and directly testable.

2. Concurrent coverage — two new tests:

• testRegisterObservesDirectlyDeliveredCancel deterministically replays your interleaving (register → full receiveRequestCancel runs and delivers the cancel directly → the registering side resumes). Verified it fails against the previous logic (register() returned true) and passes with the fix.
• testConcurrentRegisterAndCancel — 1000 rounds, two threads on a CyclicBarrier racing register() against receiveRequestCancel(id); after each round it asserts the cancel was delivered, tryCancel() ran exactly once, and the request did not stay registered.

One note on the stress test's assertions: "register() returned true for an already-cancelled request" is not black-box assertable post-hoc — a cancel landing immediately after register() returns is legal and indistinguishable from outside the group. That is why the regression assertion lives in the deterministic replay, while the stress test checks the invariants that are sound across all interleavings (no lost cancel, no double delivery, no leaked registration).

3. Parked-cancel TTL — documented on pendingCancels: a cancel that loses the race with completion parks unconditionally and the entry lives until the next purge; harmless and self-purging.

4. receiveRequest semantic change — added a Compatibility note section to the PR description and a constructor javadoc warning that subclasses must now call register() explicitly.

A side effect worth flagging: TestLocalRequest's own isCancelled() override had to go — it would clash with the new final method on the base class (and the base implementation, set before tryCancel() runs, is what the assertions want anyway).