Module: InstantRecord::Client

Defined in:
lib/instant_record/client.rb,
lib/instant_record/client/notifier.rb,
lib/instant_record/client/transport.rb

Overview

Browser-side sync client. Ruby owns the whole sync loop — transport included — via JS fetch interop; the service worker shim only schedules InstantRecord.tick (wasm Ruby cannot sleep without blocking the VM).

Defined Under Namespace

Modules: Notifier, Transport

Constant Summary collapse

BOOKKEEPING_COLUMNS =

Columns the gem keeps for its own bookkeeping. A remote value that differs in these alone is not a change to the row, it is our own write being acknowledged twice.

%w[server_version sync_state].freeze

Class Attribute Summary collapse

Class Method Summary collapse

Class Attribute Details

.notifier ⇒ Object 



34
35
36
 
# File 'lib/instant_record/client.rb', line 34

def notifier
  @notifier ||= Notifier::JsClients.new
end

.streaming=(value) ⇒ Object (writeonly)

Set while something outside the VM holds a tailing stream and feeds events in through apply_events (the service worker does this). Ruby cannot hold that stream itself: each chunk it awaited would suspend the single-flight tick, starving outbox drains for the length of the window.



134
135
136
 
# File 'lib/instant_record/client.rb', line 134

def streaming=(value)
  @streaming = value
end

.transport ⇒ Object 



30
31
32
 
# File 'lib/instant_record/client.rb', line 30

def transport
  @transport ||= Transport::JsFetch.new
end

Class Method Details

.apply_change(event) ⇒ Object

Apply one remote change event (last-write-wins). Returns whether it actually changed anything here — see upsert_from_server.



294
295
296
297
298
299
300
301
302
303
304
305
306
 
# File 'lib/instant_record/client.rb', line 294

def apply_change(event)
  event = event.deep_stringify_keys
  model = InstantRecord.synced_model(event["type"])
  return false unless model

  applying_remote do
    if event["operation"] == "destroy"
      !!model.find_by(id: event["id"])&.destroy!
    else
      upsert_from_server(model, event["attributes"], event["version"])
    end
  end
end

.apply_events(events) ⇒ Object

Applies events delivered from outside — a stream the worker is holding. Idempotent by the same upsert + last-write-wins path a poll uses, so a replayed or overlapping batch is harmless. Returns the number applied.



164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
 
# File 'lib/instant_record/client.rb', line 164

def apply_events(events)
  applied = 0
  newest_cursor = nil

  Array(events).each do |event|
    event = event.deep_stringify_keys
    applied += 1 if apply_change(event)
    # The cursor advances for every event, changed or not: an echo we chose
    # to ignore must not be handed to us again.
    newest_cursor = event["cursor"] || newest_cursor
  end

  # One cursor write per batch, as poll_changes does: a crash before this
  # re-applies the batch, which the upserts make safe.
  self.cursor = newest_cursor if newest_cursor
  notifier.records_changed if applied.positive?
  applied
end

.apply_results(results) ⇒ Object

Server results for a posted batch. Every result drops the outbox row — each one is a durable resolution, not a retry — and what happens to the record depends on which resolution it is: applied -> mark the record synced applied + skipped -> the write lost last-write-wins; reconcile to the value that won, because it did NOT land here rejected -> reconcile to server state



261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
 
# File 'lib/instant_record/client.rb', line 261

def apply_results(results)
  Array(results).each do |result|
    result = result.deep_stringify_keys
    mutation = OutboxMutation.find_by(id: result["mutation_id"])
    next unless mutation

    applying_remote do
      case result["status"]
      when "applied"
        # `skipped` is the server saying it threw this write away to
        # last-write-wins. It is resolved (nothing will retry it) but it
        # never landed, so marking the record synced would badge a
        # discarded write as delivered — the record does not hold what the
        # server holds. Reconcile to the winner, which rides along with the
        # result; the same value also arrives over the change stream, which
        # is what makes doing nothing here look fine until the stream is
        # behind.
        if result["skipped"]
          reconcile_to_server_state(mutation, result)
        else
          mark_synced(mutation, result["version"])
        end
      when "rejected"
        reconcile_rejection(mutation, result)
      end
      mutation.destroy!
    end
  end
  nil
end

.applying_remote(&block) ⇒ Object 



19
20
21
22
23
24
 
# File 'lib/instant_record/client.rb', line 19

def applying_remote(&block)
  @applying_remote = true
  block.call
ensure
  @applying_remote = false
end

.applying_remote? ⇒ Boolean

Applying remote state must never enqueue new outbox mutations.

Returns:

  • (Boolean) 


18
 
# File 'lib/instant_record/client.rb', line 18

def applying_remote? = @applying_remote

.bootstrap ⇒ Object

First-sync hydration: windowed current state + cursor in one response. The server captures the cursor BEFORE reading rows, so any overlap with subsequent events re-applies idempotently. Returns true when the snapshot applied.



103
104
105
106
107
108
109
110
111
112
113
114
115
 
# File 'lib/instant_record/client.rb', line 103

def bootstrap
  body = transport.get_json("/bootstrap")
  Array(body["records"]).each { |record| apply_change(record) }

  # Cursor written last: a crash mid-apply leaves it nil, so the next
  # tick re-bootstraps; upserts make the retry idempotent. A failure
  # never falls through to a cursor-0 poll of the full change log.
  self.cursor = body["cursor"]
  true
rescue Transport::Error => e
  log_failure("bootstrap", e)
  false
end

.bootstrapped? ⇒ Boolean

nil-aware on purpose: cursor 0 (bootstrapped against an empty change log) and "never synced" must not be conflated — cursor alone would report 0 for both.

Returns:

  • (Boolean) 


95
96
97
 
# File 'lib/instant_record/client.rb', line 95

def bootstrapped?
  !SyncMetadata.get("cursor").nil?
end

.consume_stream(chunk) ⇒ Object

Feeds one chunk of the change stream through the same SseParser the poll path uses, and applies whatever complete events it yields. The worker owns the socket — Ruby awaiting chunks would suspend the sync guard — but the wire format is parsed here, so there is one implementation of it and it is the tested one. A frame split across chunks is buffered until it is whole. Returns the number applied.



144
145
146
147
148
149
150
151
 
# File 'lib/instant_record/client.rb', line 144

def consume_stream(chunk)
  stream_parser.feed(chunk.to_s)
  return 0 if stream_events.empty?

  applied = apply_events(stream_events)
  stream_events.clear
  applied
end

.cursor ⇒ Object 



84
85
86
 
# File 'lib/instant_record/client.rb', line 84

def cursor
  SyncMetadata.get("cursor").to_i
end

.cursor=(value) ⇒ Object 



88
89
90
 
# File 'lib/instant_record/client.rb', line 88

def cursor=(value)
  SyncMetadata.set("cursor", value)
end

.drain ⇒ Object

Outbox up. Returns true when anything was sent and applied.



118
119
120
121
122
123
124
125
126
127
128
 
# File 'lib/instant_record/client.rb', line 118

def drain
  mutations = pending_mutations
  return false if mutations.empty?

  body = transport.post_json("/mutations", { mutations: mutations })
  apply_results(body["results"])
  true
rescue Transport::Error => e
  log_failure("drain", e)
  false
end

.evict_beyond_windows ⇒ Object

Trim every windowed model back to its declared window. delete_all under applying_remote: no callbacks, no outbox rows. Pending rows are excluded in the WHERE — an unsynced write is never evicted. Returns true when anything was deleted.



60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
 
# File 'lib/instant_record/client.rb', line 60

def evict_beyond_windows
  return false unless @eviction_pending

  @eviction_pending = false
  evicted = false
  InstantRecord.syncable_models.each do |model|
    window = model.instant_record_sync_window
    next unless window

    applying_remote do
      evicted = true if window.beyond_window(model.where.not(sync_state: "pending")).delete_all.positive?
    end
  end
  evicted
end

.fetch_history(model, before:, partition: nil, limit: nil) ⇒ Object

On-demand history page for a windowed model. Serves from the local database when the requested page is known-contiguous (see the per-partition low-water mark below); otherwise fetches a keyset page from the server and applies it under applying_remote — idempotent upserts, no outbox rows, no records_changed (the requesting page refreshes itself). Returns applied:, has_more: or an error hash. Callers enter through InstantRecord.fetch_history, which shares the tick's single-flight guard.



220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
 
# File 'lib/instant_record/client.rb', line 220

def fetch_history(model, before:, partition: nil, limit: nil)
  window = model.instant_record_sync_window
  raise ArgumentError, "#{model.name} declares no sync_window; fetch_history needs one" unless window
  raise ArgumentError, "#{model.name} windows by #{window.partition_by}; pass partition:" if window.partition_by && partition.nil?

  limit ||= window.limit
  before_at = Time.parse(before[:created_at].to_s)
  before_id = before[:id].to_s

  unless InstantRecord.browser?
    return { ok: true, applied: 0, has_more: local_page(window, partition, before_at, before_id, limit + 1).size > limit }
  end

  if (mark = history_mark(model, partition)) && page_local?(mark, window, partition, before_at, before_id, limit)
    # has_more from the served page, not the partition-wide mark: rows
    # remain below this page when the local probe overflows `limit`, or
    # when the page reaches the contiguity frontier and the server still
    # has older rows (mark[:has_more]). Echoing the mark alone reports
    # "beginning reached" for any mid-history page after the true
    # beginning was loaded.
    local = local_page(window, partition, before_at, before_id, limit + 1)
    return { ok: true, applied: 0, has_more: local.size > limit || mark[:has_more] }
  end

  body = transport.get_json(history_path(model, partition, before_at, before_id, limit))
  records = Array(body["records"])
  records.each { |record| apply_change(record) }
  extend_history_mark(model, partition, records, has_more: !!body["has_more"])
  { ok: true, applied: records.size, has_more: !!body["has_more"] }
rescue Transport::Error => e
  log_failure("fetch_history", e)
  { ok: false, applied: 0, has_more: nil, error: e.message }
end

.pending_count ⇒ Object 



76
77
78
 
# File 'lib/instant_record/client.rb', line 76

def pending_count
  OutboxMutation.count
end

.pending_mutations ⇒ Object 



80
81
82
 
# File 'lib/instant_record/client.rb', line 80

def pending_mutations
  OutboxMutation.ordered.map(&:as_mutation)
end

.poll_changes ⇒ Object

Changes down, catch-up style: window=0 asks the server to close the stream right after catch-up instead of tailing a long window. The tick cadence provides liveness; a long-held stream inside the single-flight tick would starve outbox drains and defer notifications.

Skipped entirely while a stream is live — that stream is already delivering, and re-polling would only re-fetch what it just applied. Returns true when any remote change was applied.



191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
 
# File 'lib/instant_record/client.rb', line 191

def poll_changes
  return false if streaming?

  changed = false
  last_cursor = nil

  transport.each_event("/events?after=#{cursor}&window=0") do |event|
    event = event.deep_stringify_keys
    changed = true if apply_change(event)
    last_cursor = event["cursor"] || last_cursor
  end

  # One cursor write per poll, not per event. A crash before this line
  # re-applies the batch next poll; upserts + LWW make that idempotent.
  self.cursor = last_cursor if last_cursor
  changed
rescue Transport::Error => e
  log_failure("poll", e)
  changed
end

.request_eviction ⇒ Object

Arm the boot-time trim; InstantRecord.start sets this on cold boots only. Runs once, on the next sync pass.



52
53
54
 
# File 'lib/instant_record/client.rb', line 52

def request_eviction
  @eviction_pending = true
end

.reset_stream ⇒ Object

A reconnect starts a new stream: drop any half-received frame from the old one rather than letting its prefix corrupt the first new frame.



155
156
157
158
159
 
# File 'lib/instant_record/client.rb', line 155

def reset_stream
  @stream_parser = nil
  stream_events.clear
  nil
end

.streaming? ⇒ Boolean

Returns:

  • (Boolean) 


136
 
# File 'lib/instant_record/client.rb', line 136

def streaming? = !!@streaming

.sync_pass ⇒ Object

One full pass: outbox up, changes down, tabs notified when anything moved. Called under InstantRecord.tick's single-flight guard. A client that has never synced hydrates from the bootstrap snapshot instead of replaying the whole change log from cursor 0.



42
43
44
45
46
47
48
 
# File 'lib/instant_record/client.rb', line 42

def sync_pass
  changed = evict_beyond_windows
  changed = drain || changed
  changed = (bootstrapped? ? poll_changes : bootstrap) || changed
  notifier.records_changed if changed
  changed
end