View Source Runbox.StateStore.RunState (runbox v21.2.0)

Module defines struct which represents state of all stateful components of the run runtime.

Run state handles savepoint management, first savepoint is created when initialized. Timestamps of savepoints are being calculated via Runbox.StateStore.ScheduleUtils.

Savepoints timestamps are calculated as follows:

run state is initialized in timestamp = 10_000 with schedule 60_000
1st savepoint timestamp is 60_000
2nd savepoint timestamp is 120_000
... and so on ...

Main purpose of run state is to save entity state. Entity state is saved to all savepoints whose timestamp is lower than or equal to entity state timestamp and state of this entity hasn't been saved to particular savepoint yet.

Entity save works as follows:

when entity state is saved in time range <60_000;120_000) and savepoint with timestamp 60_000 is the only one incomplete -> entity is saved to 1st savepoint
when entity state is saved in time range <120_000;180_000) and savepoints with timestamp 60_000 and 120_000 are incomplete -> entity is saved to both 1st and 2nd savepoint

Summary

Types

t()

Functions

init(state_store_pid, schedule, timestamp, entity_defs)

Initializes new run state.

registered_entities_count(run_state)

Returns count of registered entities in current scheduled savepoint.

save(run_state, entity_timestamp, entity_id, entity_state)

Saves entity state into all related savepoints.

Types

t()

@type t() :: %Runbox.StateStore.RunState{
  latest_collected_savepoint: Runbox.StateStore.ScheduleUtils.epoch_ms(),
  savepoints: %{
    required(Runbox.StateStore.ScheduleUtils.epoch_ms()) =>
      Runbox.StateStore.Savepoint.t()
  },
  schedule: Runbox.StateStore.ScheduleUtils.schedule()
}

Functions

init(state_store_pid, schedule, timestamp, entity_defs)

@spec init(
  pid(),
  Runbox.StateStore.ScheduleUtils.schedule(),
  Runbox.StateStore.ScheduleUtils.epoch_ms(),
  [{Runbox.StateStore.Entity.id(), Runbox.StateStore.Entity.state()}]
) :: {:ok, [Runbox.StateStore.Entity.t()], t()}

Initializes new run state.

registered_entities_count(run_state)

@spec registered_entities_count(t()) :: integer()

Returns count of registered entities in current scheduled savepoint.

save(run_state, entity_timestamp, entity_id, entity_state)

@spec save(
  t(),
  Runbox.StateStore.ScheduleUtils.epoch_ms(),
  Runbox.StateStore.Entity.id(),
  Runbox.StateStore.Entity.state()
) :: {:ok, t()} | {:state_complete, Runbox.StateStore.Savepoint.t(), t()}

Saves entity state into all related savepoints.

When savepoint is completed (all registered entities has saved state), this savepoint is returned.

Entity state is saved to all savepoints which timestamp is lower or equal than timestamp of entity state and that savepoint hasn't collected state of given entity yet.

Entity save works as follows:

when entity state is saved in time range <60_000;120_000) and savepoint with timestamp 60_000 is the only one incomplete -> entity is saved to 1st savepoint
when entity state is saved in time range <120_000;180_000) and savepoints with timestamp 60_000 and 120_000 are incomplete -> entity is saved to both 1st and 2nd savepoint