Runbox.StateStore.Storage behaviour (runbox v26.0.0)

View Source

Behaviour for generic state store storage.

Behaviour defines callbacks to handle savepoint persistence.

Summary

Types

config()
run_id()

Callbacks

handle_delete_old_savepoints(storage_state, run_id, keep_n)

Deletes old savepoints of given run but keeps n latest savepoints.

handle_delete_run(storage_state, run_id)

Deletes storage of run identified by run_id.

handle_get_latest_savepoint(storage_state, run_id)

Returns latest persisted savepoint of given run.

handle_get_latest_savepoint_timestamp(storage_state, run_id)

Returns latest persisted savepoint timestamp of given run.

handle_init_storage(storage_state, run_id)

Initializes state store for given run.

handle_initialized?(storage_state, run_id)

Predicate returning true when state store of given run has been initialized.

handle_save_savepoint(storage_state, run_id, epoch_ms, list)

Persists run state (savepoint) of given run.

Functions

delete_old_savepoints(run_id, keep_n \\ 1)

Deletes old savepoints of given run but keeps n latest savepoints.

delete_run(run_id)

Deletes storage of run identified by run_id.

get_config()
get_latest_savepoint(run_id)

Returns latest persisted savepoint of given run.

init_config(config)

Initialize state store config.

init_storage(run_id)

Initializes state store for given run.

initialized?(run_id)

Predicate returning true when state store of given run has been initialized.

last_persisted_timestamp(run_id)

Returns latest persisted savepoint timestamp of given run.

save_run_state(run_id, timestamp, entities)

Persists run state (savepoint) of given run.

Types

config()

@type config() :: {module(), term()}

run_id()

@type run_id() :: String.t()

Callbacks

handle_delete_old_savepoints(storage_state, run_id, keep_n)

@callback handle_delete_old_savepoints(
  storage_state :: term(),
  run_id(),
  keep_n :: integer()
) :: :ok

Deletes old savepoints of given run but keeps n latest savepoints.

Arguments

  • storage_state - behaviour implementation state
  • run_id
  • keep_n - count of persisted savepoints which should remain undeleted

handle_delete_run(storage_state, run_id)

@callback handle_delete_run(storage_state :: term(), run_id :: run_id()) ::
  :ok | {:error, reason :: term()}

Deletes storage of run identified by run_id.

Arguments

  • storage_state - behaviour implementation state
  • run_id

handle_get_latest_savepoint(storage_state, run_id)

@callback handle_get_latest_savepoint(storage_state :: term(), run_id()) ::
  {:ok, Runbox.StateStore.ScheduleUtils.epoch_ms(),
   [{Runbox.StateStore.Entity.id(), Runbox.StateStore.Entity.state()}]}
  | {:error, term()}

Returns latest persisted savepoint of given run.

Arguments

  • storage_state - behaviour implementation state
  • run_id

handle_get_latest_savepoint_timestamp(storage_state, run_id)

@callback handle_get_latest_savepoint_timestamp(storage_state :: term(), run_id()) ::
  {:ok, Runbox.StateStore.ScheduleUtils.epoch_ms()} | {:error, reason :: term()}

Returns latest persisted savepoint timestamp of given run.

Arguments

  • storage_state - behaviour implementation state
  • run_id

handle_init_storage(storage_state, run_id)

@callback handle_init_storage(storage_state :: term(), run_id()) :: :ok

Initializes state store for given run.

Arguments

  • storage_state - behaviour implementation state
  • run_id

handle_initialized?(storage_state, run_id)

@callback handle_initialized?(storage_state :: term(), run_id()) :: boolean()

Predicate returning true when state store of given run has been initialized.

Arguments

  • storage_state - behaviour implementation state
  • run_id

handle_save_savepoint(storage_state, run_id, epoch_ms, list)

@callback handle_save_savepoint(
  storage_state :: term(),
  run_id(),
  Runbox.StateStore.ScheduleUtils.epoch_ms(),
  [{Runbox.StateStore.Entity.id(), Runbox.StateStore.Entity.state()}]
) :: :ok

Persists run state (savepoint) of given run.

Arguments

  • storage_state - behaviour implementation state
  • run_id
  • timestamp - timestamp of given savepoint
  • entities - savepoint state representation as {entity_id, entity_state}

Functions

delete_old_savepoints(run_id, keep_n \\ 1)

@spec delete_old_savepoints(run_id(), keep_n :: integer()) :: :ok

Deletes old savepoints of given run but keeps n latest savepoints.

Arguments

  • run_id
  • keep_n - count of persisted savepoints which should remain undeleted

delete_run(run_id)

@spec delete_run(run_id()) :: :ok | {:error, reason :: term()}

Deletes storage of run identified by run_id.

Arguments

  • run_id

get_config()

@spec get_config() :: config()

get_latest_savepoint(run_id)

@spec get_latest_savepoint(run_id()) ::
  {:ok, Runbox.StateStore.ScheduleUtils.epoch_ms(),
   [{Runbox.StateStore.Entity.id(), Runbox.StateStore.Entity.state()}]}
  | {:error, :no_savepoint}

Returns latest persisted savepoint of given run.

Arguments

  • run_id

init_config(config)

@spec init_config(config()) :: :ok

Initialize state store config.

We store StateStore configuration in ETS table, because current API defines both per running run functions and global StateStore functions.

init_storage(run_id)

@spec init_storage(run_id()) :: :ok

Initializes state store for given run.

Arguments

  • run_id

initialized?(run_id)

@spec initialized?(run_id()) :: boolean()

Predicate returning true when state store of given run has been initialized.

Arguments

  • run_id

last_persisted_timestamp(run_id)

@spec last_persisted_timestamp(run_id()) ::
  {:ok, Runbox.StateStore.ScheduleUtils.epoch_ms()} | {:error, term()}

Returns latest persisted savepoint timestamp of given run.

Arguments

  • run_id

save_run_state(run_id, timestamp, entities)

@spec save_run_state(run_id(), Runbox.StateStore.ScheduleUtils.epoch_ms(), [
  {Runbox.StateStore.Entity.id(), Runbox.StateStore.Entity.state()}
]) :: :ok

Persists run state (savepoint) of given run.

Arguments

  • run_id
  • timestamp - timestamp of given savepoint
  • entities - savepoint state representation as {entity_id, entity_state}