Skip to content

ArcpClient.subscribe leaks per-call resources and double-subscribes on repeat calls for the same job #59

Description

@nficano

The subscribe method in arcp-client/src/main/java/dev/arcp/client/ArcpClient.java around line 169 creates the per-job SubmissionPublisher via computeIfAbsent and then sends a JobSubscribe envelope to the runtime — but the JOB_SUBSCRIBE send happens unconditionally on every call, not only on the first. A caller that calls subscribe(jobId, opts) twice for the same job receives the same publisher instance (good) but the runtime receives two subscribe envelopes. The runtime's SessionLoop.handleSubscribe in arcp-runtime/src/main/java/dev/arcp/runtime/session/SessionLoop.java around line 719 adds a fresh Subscriber to rec.subscribers() each time, so the second subscribe causes every subsequent job event to fire sub.session().sendJobMessage(...) twice — the consumer sees duplicated events. There is also no way to unsubscribe locally without also unsubscribing on the runtime, which JobUnsubscribe does not address (it removes the server-side mapping but leaves the local SubmissionPublisher alive).\n\nFix prompt: In arcp-client/src/main/java/dev/arcp/client/ArcpClient.java rewrite subscribe so the JOB_SUBSCRIBE wire send only fires when computeIfAbsent actually inserted a new publisher — capture that via an AtomicBoolean updated inside the computeIfAbsent lambda, or pre-check with putIfAbsent and only send when the prior was null. While there, expose a public unsubscribe(JobId) that closes the local SubmissionPublisher and sends JobUnsubscribe to the runtime. In arcp-runtime/src/main/java/dev/arcp/runtime/session/SessionLoop.java make handleSubscribe idempotent: before adding a Subscriber, check whether one already exists for this jobId/session pair, and if so refresh its lastSeenSeq instead of adding a duplicate. Add JUnit tests in arcp-client/src/test covering: two calls to subscribe(jobId, opts) return identical publishers and the consumer sees each event exactly once; unsubscribe closes the local publisher and stops further deliveries.

Activity

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Metadata

Metadata

Assignees

No one assigned

    Labels

    bugSomething isn't workingseverity:mediumMedium severity

    Type

    No type

    Projects

    No projects

      Milestone

      No milestone

      Relationships

      None yet

      Development

      No branches or pull requests

      Issue actions