Module: Y::ActionCable::Sync

Defined in:
lib/y/action_cable/sync.rb

Overview

y-websocket protocol over ActionCable.

Include this module in an ActionCable channel to sync Y.js documents (and awareness/presence) with browser clients. Messages are the standard y-protocols binary messages, base64-encoded in a JSON envelope:

{ "update" => "<base64 bytes>", "id" => 42 } # client -> server
{ "update" => "<base64 bytes>" }             # server -> subscribers
{ "ack" => 42 }                              # server -> sender

Example:

class DocumentChannel < ApplicationCable::Channel
include Y::ActionCable

on_load { |key| Document.find_by(key: key)&.content }
# on_change runs in the channel instance's context, so instance methods
# (current_user, params, ...) are available:
on_change { |key, update| Document.record!(key, update, by: current_user) }

def subscribed
  sync_subscribed params[:id]
end

def receive(data)
  sync_receive(data)
end
end

There is no unsubscribe hook: the server keeps no per-connection document or presence state, so a disconnect needs no server-side cleanup.

The concern is store-backed: every document update is recorded through on_change before it is broadcast or acked, and documents rebuild from on_load whenever state is served. No authoritative document state is kept in ActionCable process memory.

Defined Under Namespace

Modules: ClassMethods

Constant Summary collapse

MSG_KIND_SYNC_STEP1 =

Frame kinds we act on, from Y.message_kind. Its other codes (0 for a drop: malformed/truncated/multi-message/unknown, and 4 for an awareness query) fall through to a no-op in the dispatch below.

1
MSG_KIND_UPDATE = 
2
MSG_KIND_AWARENESS = 
3
DEFAULT_MAX_FRAME_BYTES =

Default incoming-frame size cap (decoded bytes). Generous enough for a large initial SyncStep2, small enough to bound a single message's allocation/parse cost. Override per channel with max_frame_bytes.

8 * 1024 * 1024

Class Method Summary collapse

Instance Method Summary collapse

Class Method Details

.included(base) ⇒ Object 



55
56
57
 
# File 'lib/y/action_cable/sync.rb', line 55

def self.included(base)
  base.extend(ClassMethods)
end

Instance Method Details

#sync_receive(data, key = nil) ⇒ Object

Call from receive. Applies the client's message, replies directly when the protocol calls for it, and relays document/awareness changes to the other subscribers.

Reliable delivery: document updates carry an "id", and the server replies { "ack" => id } once the update has been durably recorded. A causally- incomplete update is recorded and acked like any other; it stays pending (durable, invisible in the document) until its missing dependency arrives.



141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
 
# File 'lib/y/action_cable/sync.rb', line 141

def sync_receive(data, key = nil)
  # Pass `key` (params[:id]) when your transport doesn't keep the channel
  # instance alive across actions. Under AnyCable each RPC command gets a
  # fresh channel, so instance variables set in `subscribed` are gone here.
  @sync_key = key.to_s if key

  encoded = data.is_a?(Hash) ? data["update"] : nil
  return unless encoded.is_a?(String)

  # Optional client-supplied id for reliable delivery (see sync_send_ack).
  # data is known to be a Hash here (encoded came from it above).
  id = data["id"]

  # Frame-size cap: drop oversized frames before decoding (the encoded form
  # is ~4/3 the decoded size) and again after, so a client can't force large
  # base64 decodes / native parses / merges. A dropped frame is never acked,
  # and there is no protocol NACK, so a legitimate oversized update is
  # retransmitted indefinitely. Log the drop so it is at least findable.
  cap = self.class.max_frame_bytes
  if cap && encoded.bytesize > (cap * 4 / 3) + 4
    sync_log_drop(:warn, "encoded #{encoded.bytesize}B exceeds max_frame_bytes #{cap}B", id)
    return
  end

  begin
    bytes = Base64.strict_decode64(encoded)
  rescue ArgumentError
    sync_log_drop(:debug, "not valid base64", id) # garbage or a probe, rarely a real client
    return # ignore the frame and keep the connection
  end

  if cap && bytes.bytesize > cap
    sync_log_drop(:warn, "decoded #{bytes.bytesize}B exceeds max_frame_bytes #{cap}B", id)
    return
  end

  sync_send_ack(id, sync_handle_frame(encoded, bytes))
end

#sync_subscribed(key) ⇒ Object

Call from subscribed. Streams broadcasts for this document and transmits the server's opening handshake (SyncStep1 from the store).



112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
 
# File 'lib/y/action_cable/sync.rb', line 112

def sync_subscribed(key)
  @sync_key = key.to_s
  sync_validate_required_hooks!

  # The document stream is never whisper-enabled; under AnyCable we also
  # subscribe an awareness stream with `whisper: true`, scoping the client-to-
  # client path to ephemeral presence rather than the durable document stream.
  stream_from sync_stream_name
  stream_from sync_awareness_stream_name, whisper: true if respond_to?(:whispers_to)

  # The opening handshake is also the gap-repair prompt: sending our SyncStep1
  # asks the joining client for everything beyond our integrated state, which
  # is exactly the missing dependency an open gap is waiting on. If a live
  # client has it, the join heals the gap. We only surface it (on_gap); the
  # handshake below already does the soliciting.
  doc = sync_load_doc
  sync_transmit(doc.sync_step1)
  sync_observe_gap if doc.pending?
end