Module: MCPClient::SessionPin

Included in:
JsonRpcCommon
Defined in:
lib/mcp_client/session_pin.rb

Overview

Pinning a request to the server session it belongs to: a payload whose meaning is session-scoped (task ids, input request keys) must never be written into the session that replaced the one it was built in, and the transports establish (and so may re-establish) their session inside the very request that carries it. Mixed into the JSON-RPC transports, which call #check_session_pin! immediately before the wire.

Constant Summary collapse

SESSION_PINS =

Fiber-local key of the session pins in effect (see #pinned_to_session).

:mcp_client_session_pins
WRITE_GUARDS =

Fiber-local key of the extra pre-write guards (see #guarded_writes).

:mcp_client_write_guards

Instance Method Summary collapse

Instance Method Details

#check_session_pin! ⇒ void

This method returns an undefined value.

Refuse a request whose session has ended (see #pinned_to_session), or which the caller's own guard turns down (see #guarded_writes). Transports call this as late as they can, immediately before the request goes on the wire, so nothing of an ended session is written.

Raises:



95
96
97
98
99
100
101
102
103
104
105
 
# File 'lib/mcp_client/session_pin.rb', line 95

def check_session_pin!
  Thread.current[WRITE_GUARDS]&.[](self)&.call
  pinned = Thread.current[SESSION_PINS]&.[](self)
  return if pinned.nil?

  current = respond_to?(:session_epoch) ? session_epoch : nil
  return if current.nil? || current == pinned

  raise MCPClient::Errors::SessionChangedError,
        "The server session the request belongs to ended before it was sent (session #{pinned} is over)"
end

#guarded_writes(guard) ⇒ Object

Run the block with guard called immediately before every request this thread writes through this server, at the very point the session pin is checked (see #check_session_pin!). A request whose payload a concurrent answer can invalidate — the tasks extension's task ids, which a fresh CreateTaskResult hands to a different task — is guarded there rather than before the request is built: everything the client records up to the wire is seen, so a decision taken earlier cannot leave the request going out for a task that no longer exists. Only one guard is in force per server; a nested one replaces it for the duration of its block.

Parameters:

  • guard (#call) —

    raises to refuse the write

Returns:

  • (Object) —

    the block's value



55
56
57
58
59
60
61
62
63
64
65
66
 
# File 'lib/mcp_client/session_pin.rb', line 55

def guarded_writes(guard)
  previous = Thread.current[WRITE_GUARDS]
  guards = {}.compare_by_identity
  previous&.each { |server, guarded| guards[server] = guarded }
  guards[self] = guard
  Thread.current[WRITE_GUARDS] = guards
  begin
    yield
  ensure
    Thread.current[WRITE_GUARDS] = previous
  end
end

#pinned_to_session(epoch) ⇒ Object

Run the block with every request this thread sends through this server pinned to epoch (a MCPClient::ServerBase#session_epoch reading): the transport refuses to write once that session has ended, however the reconnect that ended it got in — a lazy ensure_initialized / ensure_connected inside the very request, a transport retry, a concurrent cleanup. A session-scoped payload (the tasks extension's inputResponses, whose task ids and input keys are per session and reusable) must never reach the session that replaced the one it was built in, where it could answer an unrelated request.

Parameters:

  • epoch (Integer, nil) —

    the session the requests belong to (nil: no pin)

Returns:

  • (Object) —

    the block's value



29
30
31
32
33
34
35
36
37
38
39
40
41
42
 
# File 'lib/mcp_client/session_pin.rb', line 29

def pinned_to_session(epoch)
  return yield if epoch.nil?

  previous = Thread.current[SESSION_PINS]
  pins = {}.compare_by_identity
  previous&.each { |server, pinned| pins[server] = pinned }
  pins[self] = epoch
  Thread.current[SESSION_PINS] = pins
  begin
    yield
  ensure
    Thread.current[SESSION_PINS] = previous
  end
end

#unpinned_session ⇒ Object

Run the block with this server's pin — and its pre-write guard — lifted for this thread: the request that establishes the session replacing an ended one is not part of the session it replaces, and the pin (whose epoch the end of that session has just invalidated) would otherwise refuse the very handshake the caller is in the middle of performing.

Returns:

  • (Object) —

    the block's value



74
75
76
77
78
79
80
81
82
83
84
85
86
87
 
# File 'lib/mcp_client/session_pin.rb', line 74

def unpinned_session
  previous = Thread.current[SESSION_PINS]
  guarded = Thread.current[WRITE_GUARDS]
  return yield if (previous.nil? || !previous.key?(self)) && (guarded.nil? || !guarded.key?(self))

  Thread.current[SESSION_PINS] = without_self(previous)
  Thread.current[WRITE_GUARDS] = without_self(guarded)
  begin
    yield
  ensure
    Thread.current[SESSION_PINS] = previous
    Thread.current[WRITE_GUARDS] = guarded
  end
end