CrowdControl.Store behaviour (crowd_control v0.1.1)

Copy Markdown View Source

Behaviour for persisting enough session state to reattach after a restart.

A remote sandbox outlives the CrowdControl.Session that created it. When the VM restarts, the only way to find those sandboxes again — and to resume reading their output without losing or duplicating a byte — is a record written before the crash. That is what a store holds.

Two implementations ship, neither adding a dependency:

Callers wanting cross-node durability implement this behaviour over Ecto, Redis, or anything else; it is four functions.

Configuration

config :crowd_control, :store, CrowdControl.Store.ETS

config :crowd_control, :store, {CrowdControl.Store.DETS, path: "/var/lib/cc/sessions.dets"}

What is stored, and what deliberately is not

See t:record/0. The cursor halves — :byte_offset and :buffer — are the point of the whole thing: byte_offset says where to resume reading the sandbox's output file, and :buffer carries the partial line that was in-flight when the session died. Reattaching seeds the buffer and reads from the offset, so a line split across the failure is rejoined exactly.

Not stored:

  • messages / message_count — already a lossy window capped by :max_messages, and rebuildable by replaying the sandbox's output from offset 0 if a caller ever needs it
  • subscribers — pids, meaningless after a restart; callers re-subscribe/1
  • timeout_ref, reader — process-local, rebuilt on reattach

Only reattachable backends write

CrowdControl.Backend.Local cannot reattach — a local subprocess dies with the VM — so a store write per stdout chunk would be pure overhead. Session checks CrowdControl.Backend.reattachable?/1 and skips persistence entirely for such backends.

Two different ids

Records are keyed by the CrowdControl session key — a random id minted by CrowdControl.Session at startup, before the sandbox is provisioned, and stamped onto the sandbox as the crowd_control.session label. That label is what lets CrowdControl.Reaper match a running container back to its record.

The CLI's own session id (:session_id) is a different thing: it does not exist until the CLI emits system/init, and it is only useful for --resume. Keying on it would leave every session unfindable for the first few hundred milliseconds of its life — precisely the window in which a crash strands a container nobody can reap.

Summary

Types

t()

A persisted session.

Callbacks

all()

Every record currently held.

delete(key)

Remove a record. Absent is success.

get(key)

Fetch a record, or :error if absent.

put(key, t)

Insert or replace the record for key.

Functions

all()

List via the configured store.

build(fields)

Build a record from session state.

delete(key)

Delete via the configured store.

get(key)

Read via the configured store.

impl()

The configured store module.

new_key()

Mint a new session key.

owner_id()

This node's owner id.

put(key, record)

Write via the configured store.

resolve()

The configured store as {module, opts}.

scrub_opts(opts)

Remove credentials from an options keyword list.

secret_keys()

Credential-bearing option keys, stripped before anything is persisted.

Types

t()

@type t() :: %{
  :key => String.t(),
  :session_id => String.t() | nil,
  :backend => module(),
  :handle => term(),
  :byte_offset => non_neg_integer(),
  :buffer => binary(),
  :opts => keyword(),
  :owner => String.t(),
  :updated_at => integer(),
  optional(atom()) => term()
}

A persisted session.

  • :key — the CrowdControl session key; the store key and sandbox label
  • :session_id — the CLI's own session id, used for --resume. nil until the CLI emits system/init.
  • :backend — the backend module, so the reaper knows who owns the handle
  • :handle — backend-opaque; must survive :erlang.term_to_binary/1
  • :byte_offset — bytes of sandbox output already delivered to the session
  • :buffer — partial line in flight at the last write
  • :opts — the session opts, replayed on reattach
  • :owner — this node's owner id; scopes reaping (see CrowdControl.Reaper)
  • :updated_at — System.system_time(:millisecond)

Callbacks

all()

@callback all() :: [t()]

Every record currently held.

delete(key)

@callback delete(key :: String.t()) :: :ok

Remove a record. Absent is success.

get(key)

@callback get(key :: String.t()) :: {:ok, t()} | :error

Fetch a record, or :error if absent.

put(key, t)

@callback put(key :: String.t(), t()) :: :ok

Insert or replace the record for key.

Functions

all()

@spec all() :: [t()]

List via the configured store.

build(fields)

@spec build(keyword()) :: t()

Build a record from session state.

Stamps :owner and :updated_at so callers cannot forget them.

delete(key)

@spec delete(String.t()) :: :ok

Delete via the configured store.

get(key)

@spec get(String.t()) :: {:ok, t()} | :error

Read via the configured store.

impl()

@spec impl() :: module()

The configured store module.

new_key()

@spec new_key() :: String.t()

Mint a new session key.

Hex so it is safe as a container label value and a filename component.

owner_id()

@spec owner_id() :: String.t()

This node's owner id.

Every sandbox is labelled with it, and CrowdControl.Reaper only ever destroys sandboxes carrying its own. Two nodes with independent stores therefore cannot reap each other's work. Defaults to to_string(node()).

config :crowd_control, :owner_id, "prod-worker-1"

put(key, record)

@spec put(String.t(), t()) :: :ok

Write via the configured store.

resolve()

@spec resolve() :: {module(), keyword()}

The configured store as {module, opts}.

Accepts a bare module or a {module, opts} tuple, mirroring CrowdControl.Backend.resolve/1.

scrub_opts(opts)

@spec scrub_opts(keyword()) :: keyword()

Remove credentials from an options keyword list.

Store records outlive the process that wrote them, and with CrowdControl.Store.DETS they outlive the VM on disk. Nothing about reattaching a session needs its API key — the sandbox already holds whatever environment it was started with — so the key has no business being written down.

iex> CrowdControl.Store.scrub_opts(api_key: "sk-real", timeout: 5000)
[timeout: 5000]

secret_keys()

@spec secret_keys() :: [atom()]

Credential-bearing option keys, stripped before anything is persisted.

:sandboxd_secret is here even though it is normally read from application config rather than passed in opts: a caller may pass it, and a stray copy in a persisted record would defeat the entire point of deriving the agent token instead of storing it.

:gce_config holds a %GcpCompute.Config{}, which carries a live token-provider argument. It is not a secret by name, which is exactly why it needs naming here.