Scriba.Position (Scriba v0.1.0)

Copy Markdown View Source

Per-stream position tracking helpers — pure functions, no process.

Position lives in two places:

  • Postgres (scriba_positions table) — authoritative. Updated atomically with read-model writes inside the target's Ecto.Multi via multi/5. Per-stream rows: PK (projection_name, projection_version, stream_id).
  • ETS (one shared named table for the whole BEAM, keyed on {name, version, stream_id} tuples) — hot-read cache for Scriba.info/1. Not authoritative.

Shared-table model

The cache lives in a single named ETS table — Scriba.Position.Cache — that is created once by Scriba.Supervisor.init/1 and lives for the application's lifetime. Each cached entry is keyed by a {name, version, stream_id} tuple; init_cache/3 and drop_cache/2 scope all their work to one projection's rows in that shared table.

This replaces the earlier per-projection :named_table design, which generated a fresh atom per projection — unbounded growth of the BEAM atom table when projection names were not bounded (e.g. property tests with :erlang.unique_integer/1-derived names). The shared table allocates exactly one atom regardless of projection count.

No PositionStore GenServer mediates the cache. Workers (the Pipeline) write directly to the shared table via cache_put/4. Readers (Scriba.info/2, telemetry) read directly via cache_get/3 / stream_positions/2 and the internal safe-position helper.

Lifecycle

  • Scriba.Supervisor.init/1 calls create_shared_table/0 once at boot.
  • Coordinator on entering :running calls init_cache/3 — wipe-then-preload: deletes any stale rows for this projection (from a previously-crashed Coordinator), then preloads from Postgres if :repo is set. Idempotent across pause→resume cycles.
  • Coordinator on entering :stopped calls drop_cache/2 to clean up this projection's rows. (Crash recovery is covered by the wipe in the next init_cache/3; :stopped cleanup matters for projections that the user explicitly stops permanently.)

Why no schema module

We use raw SQL through Ecto.Adapters.SQL.query/3 rather than an Ecto.Schema. The position rows are internal plumbing — there's no business logic that needs a changeset, and avoiding a schema means Scriba doesn't impose primary-key or timestamp conventions on the user.

Summary

Types

name()
position()
stream_id()
version()

Functions

cache_get(name, version, stream_id)

Returns {:ok, position} for one stream from the ETS cache, or :error on cache miss.

cache_get(name, version, stream_id, opts)

Cache-first lookup with optional Postgres fallback.

cache_put(name, version, stream_id, position)

Writes one stream's position into the shared cache. Silently no-ops if the shared table doesn't exist (path used by unit-test code that bypasses the supervisor).

cache_table()

Returns the atom name of the shared ETS cache table. Useful for introspection (:ets.info/1, :observer).

create_shared_table()

Creates the shared ETS cache table. Called once from Scriba.Supervisor.init/1. Idempotent — safe to call when the table already exists (returns :ok either way).

drop_cache(name, version)

Removes all cache entries belonging to (name, version) from the shared table. Returns the number of rows deleted.

init_cache(name, version, opts \\ [])

Wipes any existing cache entries for (name, version) and (optionally) preloads up to 10000 rows from Postgres into the shared table.

multi(multi, name, version, stream_id, new_position)

Appends a per-stream position upsert to an Ecto.Multi. Use this from inside an Scriba.Target.apply_batch/6 implementation building the atomic-commit Multi alongside its handler operations.

read_from_repo(repo, name, version, stream_id)

Reads one stream's position from Postgres. Returns nil if no row exists for this (name, version, stream_id).

resolve_repo(opts, target_spec)

Resolves the Ecto.Repo to use for position-tracking I/O — both the Coordinator's init_cache/3 preload and the Pipeline's source-side dedup cache_get/4 fallback consult this.

stream_positions(name, version)

Returns a %{stream_id => position} map of all streams known to the cache for this projection. Empty map if the shared table doesn't exist or no streams are tracked.

Types

name()

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

position()

@type position() :: non_neg_integer()

stream_id()

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

version()

@type version() :: pos_integer()

Functions

cache_get(name, version, stream_id)

@spec cache_get(name(), version(), stream_id()) :: {:ok, position()} | :error

Returns {:ok, position} for one stream from the ETS cache, or :error on cache miss.

Pure ETS lookup — no I/O. If you want a cache-then-Postgres lookup that also backfills the cache on miss, use cache_get/4 with repo:.

cache_get(name, version, stream_id, opts)

@spec cache_get(name(), version(), stream_id(), keyword()) ::
  {:ok, position()} | :error

Cache-first lookup with optional Postgres fallback.

Behaviour:

  • Hit in ETS → {:ok, position} (no Postgres call).
  • Miss in ETS, no :repo opt → :error.
  • Miss in ETS, :repo opt set → reads from Postgres via read_from_repo/4. If the row exists, the cache is back-filled and {:ok, position} is returned. If the row does not exist (no commit for this stream yet), returns :error.

Source-side dedup is the primary caller of the :repo-backed form: dedup needs the durable cursor for a stream that hasn't appeared in the cache yet (e.g. a newly-discovered stream after Coordinator restart with a cache preloaded only up to @preload_cap rows).

cache_put(name, version, stream_id, position)

@spec cache_put(name(), version(), stream_id(), position()) :: :ok

Writes one stream's position into the shared cache. Silently no-ops if the shared table doesn't exist (path used by unit-test code that bypasses the supervisor).

cache_table()

@spec cache_table() :: atom()

Returns the atom name of the shared ETS cache table. Useful for introspection (:ets.info/1, :observer).

create_shared_table()

@spec create_shared_table() :: :ok

Creates the shared ETS cache table. Called once from Scriba.Supervisor.init/1. Idempotent — safe to call when the table already exists (returns :ok either way).

The supervisor process owns the table; it dies with the application.

drop_cache(name, version)

@spec drop_cache(name(), version()) :: non_neg_integer()

Removes all cache entries belonging to (name, version) from the shared table. Returns the number of rows deleted.

No-op (returns 0) if the shared table doesn't exist yet — this can happen in unit-tested code paths that bypass the application supervisor.

init_cache(name, version, opts \\ [])

@spec init_cache(name(), version(), keyword()) :: :ok

Wipes any existing cache entries for (name, version) and (optionally) preloads up to 10000 rows from Postgres into the shared table.

Pass repo: SomeRepo to enable Postgres preload. Beyond the @preload_cap cutoff, un-preloaded streams fall back to lazy lookup via cache_get/4 with repo:.

Returns :ok.

Emits [:scriba, :projection, :cache_initialized] telemetry with measurements %{wiped_count, preloaded_count} and metadata %{name, version, source} where source is :postgres (preloaded) or :empty (no repo configured).

multi(multi, name, version, stream_id, new_position)

@spec multi(Ecto.Multi.t(), name(), version(), stream_id(), position()) ::
  Ecto.Multi.t()

Appends a per-stream position upsert to an Ecto.Multi. Use this from inside an Scriba.Target.apply_batch/6 implementation building the atomic-commit Multi alongside its handler operations.

The Multi step is keyed by {:scriba_position, stream_id} so a batch touching N streams produces N independent steps.

Monotonicity is enforced here, not upstream

The upsert is GREATEST(existing, incoming), so a cursor can never move backwards no matter what the caller passes.

The Pipeline also filters :skip results out of stream_advances so a redelivered below-cursor batch does not regress the cursor. That filter is still correct and still wanted — but it is one Enum.reject/2 in another module, and a cursor moving backwards is this library's worst failure mode: it silently re-applies committed effects. The invariant belongs in the storage layer where no upstream change can violate it.

Trade-off, deliberately taken: GREATEST also masks a genuine Pipeline bug that computes a regressing advance, converting loud corruption into a silent no-op. That is the right trade for a v0.1 whose stated first principle is correctness over throughput. If detection is later wanted, add WHERE EXCLUDED.position > scriba_positions.position and raise on zero rows affected — but do not go back to an unconditional SET.

read_from_repo(repo, name, version, stream_id)

@spec read_from_repo(module(), name(), version(), stream_id()) :: position() | nil

Reads one stream's position from Postgres. Returns nil if no row exists for this (name, version, stream_id).

resolve_repo(opts, target_spec)

@spec resolve_repo(keyword(), {module(), keyword()}) :: module() | nil

Resolves the Ecto.Repo to use for position-tracking I/O — both the Coordinator's init_cache/3 preload and the Pipeline's source-side dedup cache_get/4 fallback consult this.

Single source of truth: keeping repo resolution in one place means the Coordinator and Pipeline can never disagree about which repo backs the cache. Diverging would silently break source-side dedup (Pipeline reads one repo, Coordinator writes another).

Order:

  1. Explicit :repo opt on the projection wins (override path; e.g. a custom target that also wants cache preload from Postgres).
  2. Scriba.Target.Ecto's target_spec carries :repo in its target opts — extract it. Raises if missing, since the Ecto target needs it anyway.
  3. Otherwise nil (Test target, custom non-Postgres targets).

stream_positions(name, version)

@spec stream_positions(name(), version()) :: %{required(stream_id()) => position()}

Returns a %{stream_id => position} map of all streams known to the cache for this projection. Empty map if the shared table doesn't exist or no streams are tracked.