Class: MTProto::Updates::StateMachine

Inherits:
Object
  • Object
show all
Defined in:
lib/mtproto/updates/state_machine.rb

Overview

The authoritative update state machine: owns the account position (pts/qts/date/seq) and per-channel pts, applies live updates in order, and recovers gaps via getDifference/getChannelDifference — but ONLY when a real sequence gap is detected, never on a blind timer. Because a gap-triggered getDifference returns exactly the missed range and the position is advanced atomically, the host never sees a duplicate in steady state (a crash mid-handler may re-deliver one update — the accepted at-least-once edge).

Sequential and Async-free: it issues blocking RPCs through the injected client, so it MUST be driven off the receiver fiber (run_updates feeds it from a single worker fiber, which also serializes recovery). Checkpoints run through store.transaction, so a host can commit the pts advance together with its own handling of the event.

Constant Summary collapse

RECOVERY_PAGES = 
500

Instance Method Summary collapse

Constructor Details

#initialize(client:, store:, logger: nil) ⇒ StateMachine

Returns a new instance of StateMachine.



23
24
25
26
27
 
# File 'lib/mtproto/updates/state_machine.rb', line 23

def initialize(client:, store:, logger: nil)
  @client = client
  @store = store
  @logger = logger
end

Instance Method Details

#process(constructor, body) ⇒ Object

Feed one raw live update container (constructor + body) from the push stream.



41
42
43
 
# File 'lib/mtproto/updates/state_machine.rb', line 41

def process(constructor, body, &)
  dispatch(Parser.parse_container(constructor, body), &)
end

#resume ⇒ Object 



29
30
31
32
33
34
35
36
37
38
 
# File 'lib/mtproto/updates/state_machine.rb', line 29

def resume(&)
  stored = @store.load_state
  base = @client.api.get_updates_state
  @pts = nonzero(stored && stored[:pts]) || base.pts
  @qts = nonzero(stored && stored[:qts]) || base.qts
  @date = nonzero(stored && stored[:date]) || base.date
  @seq = nonzero(stored && stored[:seq]) || base.seq
  persist_state
  catch_up(&) if stored
end

#resync ⇒ Object

A gap point the host detected out of band (notably a reconnect): re-sync now.



46
47
48
 
# File 'lib/mtproto/updates/state_machine.rb', line 46

def resync(&)
  catch_up(&)
end