Embedded Trellis for Elixir: define streaming transforms over Postgres
tables from the BEAM, through the trellis Rust crate itself
(docs/decisions/0010-embeddable-clients.md).
The surface mirrors the Rust crate's BlockingTrellis:
Lifecycle:
connect/1,migrate/1,config/1,shutdown/1, andstart_link/1for a handle a supervisor owns.Every statement form:
apply/2runs one statement of Trellis's grammar (TRANSFORM,PAUSE TRANSFORM,RESUME TRANSFORM,DROP TRANSFORM,ALTER TRANSFORM) and reports what it did as aTrellis.Applied.t/0.define/2is the one guarded convenience: anapply/2that refuses anything but aTRANSFORMstatement before applying it.Reads:
status/2,definitions/1, anddump_definitions_sql/1, the definitions as SQL to append to a schema dump.Quarantine:
quarantined/1,quarantine_status/2,sample_quarantined/3,poisoned_since/2. Pausing and resuming are statements:apply(trellis, "RESUME TRANSFORM order_totals.total").Operations:
request_backfill/2,detect_capture_disruptions/1,has_live_drain_workers/1,has_live_staging_worker/1, the read-your-writes pairwatermark_token/1andawait_converged/3, and the auditself_check/3, a background job thatself_check_job/2reads back.Observability:
Trellis.Metrics.render_prometheus/0for a scrape route the host serves, andTrellis.LogBridge, which forwards the engine's log lines toLoggerfrom the moment:trellis_pgstarts.# A deploy's migration step: the defaults run nothing in the background. {:ok, migrator} = Trellis.connect(url: "postgres://localhost/app") :ok = Trellis.migrate(migrator) :ok = Trellis.shutdown(migrator)
# The app, once migrated. The staging worker reads the catalog as it # starts, so a
staging: truehandle can't connect beforemigrate/1. {:ok, trellis} = Trellis.connect(url: "postgres://localhost/app", staging: true, drain_threads: 2) {:ok, "public.widget_prices"} = Trellis.define(trellis, "TRANSFORM widget_prices FROM widgets SELECT price AS price") {:ok, %Trellis.Status{status: :live}} = Trellis.status(trellis, "widget_prices") :ok = Trellis.shutdown(trellis)
Every function here blocks the calling process until the engine answers,
and holds no scheduler while it waits: the call is submitted and the
process waits in a receive, so a stuck call costs the process nothing but
its own wait, and killing the process ends the wait at once. Every call
returns within 30 seconds (a :timeout error if the database is too slow);
see "Calls" below. Non-bang functions
return {:ok, value} (or :ok) and {:error, %Trellis.Error{}}; the bang
variants return the value or raise the Trellis.Error.
Conventions
- Times are
DateTimes in UTC, at microsecond precision. - A transform is named by its target table, bare (
"order_totals") or qualified asdefine/2returns it ("public.order_totals"), instatus/2,quarantine_status/2,sample_quarantined/3,release_key/4,self_check/3and thePAUSE,RESUME,DROPandALTERstatements. - A quarantine target is an address string: a transform's name, or
"transform.column"("order_totals.total", always the bare transform), exactly asquarantined/1reports it. sample_quarantined/3's cursor andwatermark_token/1's token are opaque: pass back what the previous call returned.- Every atom in a result comes from a closed set allocated when the NIF loads; none is ever built from a string the database returned.
One handle per instance
An instance is one catalog schema in one database (the :schema option). A
handle belongs to one instance, and a VM holds a handle for each instance it
uses: against different databases, or against different catalog schemas in
one database. Connect each handle once, at boot, and share it; don't connect
per request. The handle is shut down when shutdown/1 is called, or, as a
backstop, when it is garbage collected.
Each handle pays for itself. It owns a connection pool of up to 20
connections and a small Rust runtime (:worker_threads threads). A handle
with staging: true or drain_threads above 0 also owns a second runtime
and pool of the same size, plus connections outside both pools: one for the
staging worker, and for each drain thread a LISTEN connection and up to two
more while it drains.
Handles share no budget, so size all of them together against the server's
max_connections. Each instance has its own staging worker, so set
staging: true in one process of the fleet per instance.
In an application, let a supervisor own each handle: {Trellis, options} in
the supervision tree starts a process that connects the handle as it starts
and shuts it down when the supervisor stops it (see start_link/1). Every
function here that takes a handle takes that process's name too, and the
names tell the handles apart:
# config/runtime.exs
config :my_app, MyApp.Trellis,
name: MyApp.Trellis,
url: System.fetch_env!("DATABASE_URL")
# lib/my_app/application.ex
children = [
MyApp.Repo,
{Trellis, Application.fetch_env!(:my_app, MyApp.Trellis)},
MyAppWeb.Endpoint
]
# Anywhere in the app.
{:ok, status} = Trellis.status(MyApp.Trellis, "widget_prices")A second instance in the same database is a second child with its own :name
and :schema:
{Trellis, name: MyApp.Billing, url: url, schema: "billing"}Calls
A call returns within 30 seconds, counted from when it is submitted: if the
database can't answer in time (a lock held by another session, say), the
server stops what the call started, it rolls back, and it returns
{:error, %Trellis.Error{code: :timeout}}. Heavy work isn't done inside a
call: define/2 and apply/2 register and return, and a caller that
needs the result polls status/2.
Calls run in parallel, on one handle and across processes: a call stuck on
a lock doesn't delay another process's status/2. A process's own calls
stay in order, because it waits for each reply. Calls from different
processes have no order. A process killed during a call leaves the call
running until it ends or reaches its deadline, and its reply is dropped.
shutdown/1 doesn't wait for calls in flight: it cancels them, and each
of their callers gets a :validation error, as a call made after the
shutdown does.
Trellis.Metrics.render_prometheus/0 covers every handle in the VM at once,
each instance's series labeled trellis_instance.
For the migrations that define transforms, see Trellis.Migration.
Summary
Types
Options for sample_quarantined/3
Options for self_check/3
Options for start_link/1: every option/0, plus :name, the name to
register the process under (see GenServer.start_link/3). Without one,
call it by its pid.
What every function but connect/1 takes: a handle connect/1 returned,
or the name or pid of a supervised Trellis (start_link/1).
An opaque watermark_token/1 token.
Functions
Runs one statement of Trellis's grammar, whatever its form, and reports
what it did (see Trellis.Applied.t/0).
Like apply/2, but raises Trellis.Error.
Waits until every change committed at or before token has reached its
target tables, or timeout_ms passes (a :timeout error; retry it).
Like await_converged/3, but raises Trellis.Error.
The child spec {Trellis, options} stands for in a supervision tree. Its
id is the :name option, so one tree can hold several.
The configuration trellis connected with. See Trellis.Config.
Like config/1, but raises Trellis.Error.
Connects to the database and starts whatever background work options
asks for. See option/0.
Like connect/1, but raises Trellis.Error.
Registers a TRANSFORM statement and creates its target table.
Like define/2, but raises Trellis.Error.
Every registered transform definition, oldest first.
Like definitions/1, but raises Trellis.Error.
Pauses, now, the definitions that read a table whose Trellis capture was changed outside Trellis (a trigger disabled, dropped or recreated, a capture function replaced, the table gone), and returns the qualified names of those tables.
Like detect_capture_disruptions/1, but raises Trellis.Error.
The SQL that loads this instance's definitions into a database built from
a schema-only dump (pg_dump --schema-only, mix ecto.dump). Append it
to the dump; mix trellis.dump does.
Like dump_definitions_sql/1, but raises Trellis.Error.
Whether at least one drain worker is alive anywhere in the fleet. With none, nothing reaches a target table: poll this from a health check.
Like has_live_drain_workers/1, but raises Trellis.Error.
Whether the staging worker (which installs change capture and starts backfills) is alive anywhere in the fleet. The other half of the health check.
Like has_live_staging_worker/1, but raises Trellis.Error.
Creates or upgrades Trellis's own tables in the configured schema. Safe to run on every boot.
Like migrate/1, but raises Trellis.Error.
Every source row the apply path poisoned (gave up on) after since,
oldest first. Poll it with the last entry's poisoned_at so a
whole-table failure doesn't sit unnoticed.
Like poisoned_since/2, but raises Trellis.Error.
The state of one target: a transform ("order_totals") or one of its
columns ("order_totals.total"). A column that isn't paused is :live.
Like quarantine_status/2, but raises Trellis.Error.
Every quarantined transform and paused column, across every transform. Cheap enough for a dashboard or health check to poll.
Like quarantined/1, but raises Trellis.Error.
Releases one key transform holds in quarantine, once its cause is fixed.
source_table ("public.orders" or "orders") and key are as
sample_quarantined/3 and poisoned_since/2 report them.
Like release_key/4, but raises Trellis.Error.
Rebuilds every definition that reads source_table (a bare table name,
resolved like one in a statement), and returns once that rebuild is
queued.
Like request_backfill/2, but raises Trellis.Error.
One page of the rows quarantined under target, to diagnose a
quarantine's cause: for a column ("order_totals.total"), the rows that
failed evaluating it; for a whole transform ("order_totals"), the keys
poisoned from its source table.
Like sample_quarantined/3, but raises Trellis.Error.
Starts a background check of target_table against a fresh recompute of
its definition from the source tables, and returns the Trellis.SelfCheckJob
at once. See self_check_option/0.
Like self_check/3, but raises Trellis.Error.
The Trellis.SelfCheckJob that self_check/3 returned with this id, as
it stands now: its state, how many keys it has compared so far, and, once
it is :done, its report. nil when there is none: a newer self_check/3
of its target replaced it, or its transform was dropped.
Like self_check_job/2, but raises Trellis.Error.
Stops the handle's background work and waits for its threads to exit. Calls
still in flight are cancelled, not waited for: their callers get a
:validation error, like any later call on the handle. Shutting down again
is :ok.
Like shutdown/1, but raises Trellis.Error.
Starts a process that owns a handle: it connects with options (see
start_option/0) as it starts, and shuts the handle down as its
supervisor stops it. Put it in a supervision tree as {Trellis, options}
rather than calling this directly.
Like status/2, but raises Trellis.Error.
A read-your-writes token covering every write committed before this call.
Take it after a source-table write commits, then pass it to
await_converged/3 to wait for that write to reach its targets.
Like watermark_token/1, but raises Trellis.Error.
Types
@type option() :: {:url, String.t()} | {:schema, String.t()} | {:target_schema, String.t()} | {:staging, boolean()} | {:drain_threads, non_neg_integer()} | {:worker_threads, pos_integer()}
Options for connect/1:
:url(required): the database, as apostgres://URL or a libpqkey=valuestring. Nothing is read from the environment.:schema: the schema Trellis keeps its own tables in. Default"trellis". Not"public", not:target_schema, and not another instance's catalog or target schema;migrate/1refuses those.:target_schema: the schema a bare target table name is created in. Default"public".:staging: whether this connection runs the staging worker, which installs change capture on the source tables and starts each new transform's backfill. Exactly one connection of each instance in a fleet should. Defaultfalse.:drain_threads: how many threads apply staged changes to the targets. Default0.:worker_threads: the worker threads of each Rust runtime a handle owns (its calls', and its background client's). Default2; the work is IO, so a small count is enough, and the BEAM has already sized its own schedulers to the cores.
The defaults (staging: false, drain_threads: 0) run nothing in the
background, which is right for a web process or a migration run. Some
connection in the fleet has to set both, or no definition ever leaves
:waiting_to_backfill.
@type sample_option() :: {:limit, pos_integer()} | {:after, Trellis.SamplePage.cursor() | nil}
Options for sample_quarantined/3:
:limit: the most rows to return. Default100.:after: thenext_cursorof the previous page. Defaultnil, the first page.
@type self_check_option() :: {:timeout_ms, non_neg_integer()} | {:mode, :standard | :strict}
Options for self_check/3:
:timeout_ms(required): how long a page of the job waits for the target to catch up, per wait. A:standardcheck waits up to twice per page, a:strictone once.:mode::standard(default) re-checks anything that differs after a fresh wait, so a change still in flight isn't reported; it is safe while the source is being written.:strictskips the re-check, and is only sound once writes to the audited tables have stopped.
@type start_option() :: option() | {:name, GenServer.name()}
Options for start_link/1: every option/0, plus :name, the name to
register the process under (see GenServer.start_link/3). Without one,
call it by its pid.
@opaque t()
@type trellis() :: t() | GenServer.server()
What every function but connect/1 takes: a handle connect/1 returned,
or the name or pid of a supervised Trellis (start_link/1).
@opaque watermark()
An opaque watermark_token/1 token.
Functions
@spec apply(trellis(), String.t()) :: {:ok, Trellis.Applied.t()} | {:error, Trellis.Error.t()}
Runs one statement of Trellis's grammar, whatever its form, and reports
what it did (see Trellis.Applied.t/0).
{:ok, {:transform_defined, "public.order_totals"}} =
Trellis.apply(trellis, "TRANSFORM order_totals FROM orders SELECT price + tax AS total")
{:ok, {:resumed, ["order_totals.total"]}} =
Trellis.apply(trellis, "RESUME TRANSFORM order_totals.total")A statement that doesn't parse is a :parse error, and nothing is applied.
Like define/2, this runs on Trellis's own connections, not in the
caller's transaction.
{:ok, :unknown} is a success like any other {:ok, _}: the statement
took effect, and only its outcome is newer than this binding can describe.
Don't retry it; read the result back with status/2 or definitions/1 if
you need it.
@spec apply!(trellis(), String.t()) :: Trellis.Applied.t()
Like apply/2, but raises Trellis.Error.
@spec await_converged(trellis(), watermark(), non_neg_integer()) :: :ok | {:error, Trellis.Error.t()}
Waits until every change committed at or before token has reached its
target tables, or timeout_ms passes (a :timeout error; retry it).
It waits for captured changes only: a transform that isn't :live yet
may still be missing rows when this returns (see status/2).
A call returns within 30 seconds, so a longer timeout_ms is cut short
with a :timeout error: to wait longer, call again. The wait holds no
scheduler, and other calls on the handle run beside it.
@spec await_converged!(trellis(), watermark(), non_neg_integer()) :: :ok
Like await_converged/3, but raises Trellis.Error.
@spec child_spec([start_option()]) :: Supervisor.child_spec()
The child spec {Trellis, options} stands for in a supervision tree. Its
id is the :name option, so one tree can hold several.
@spec config(trellis()) :: {:ok, Trellis.Config.t()} | {:error, Trellis.Error.t()}
The configuration trellis connected with. See Trellis.Config.
@spec config!(trellis()) :: Trellis.Config.t()
Like config/1, but raises Trellis.Error.
@spec connect([option()] | %{optional(atom()) => term()}) :: {:ok, t()} | {:error, Trellis.Error.t()}
Connects to the database and starts whatever background work options
asks for. See option/0.
Like connect/1, but raises Trellis.Error.
@spec define(trellis(), String.t()) :: {:ok, String.t()} | {:error, Trellis.Error.t()}
Registers a TRANSFORM statement and creates its target table.
Returns the transform's fully qualified target table once the definition is
registered and its backfill queued, with the transform at
:waiting_to_backfill; poll status/2 with the returned name for :live.
Only TRANSFORM statements belong here. Any other statement form (DROP,
PAUSE, ALTER, ...) is refused with a :validation error before
anything is applied, and a statement that doesn't parse is a :parse error.
apply/2 takes every form.
Trellis runs this on its own connections, not in the caller's transaction: if an enclosing Ecto migration rolls back, the definition stays.
Like define/2, but raises Trellis.Error.
@spec definitions(trellis()) :: {:ok, [Trellis.DefinitionSummary.t()]} | {:error, Trellis.Error.t()}
Every registered transform definition, oldest first.
@spec definitions!(trellis()) :: [Trellis.DefinitionSummary.t()]
Like definitions/1, but raises Trellis.Error.
@spec detect_capture_disruptions(trellis()) :: {:ok, [String.t()]} | {:error, Trellis.Error.t()}
Pauses, now, the definitions that read a table whose Trellis capture was changed outside Trellis (a trigger disabled, dropped or recreated, a capture function replaced, the table gone), and returns the qualified names of those tables.
The staging worker's reconcile pass does the same on its own cadence, with
a Trellis.CaptureFailure of kind :disruption on each paused
definition's status. This is for a caller that has just done something
that disrupts capture, like a fixture load, and needs the readers paused
before it goes on. It installs, repairs and rebuilds nothing, and resuming
is still the caller's (Trellis.TestSupport does it for a test suite).
Safe beside a running staging worker.
Like detect_capture_disruptions/1, but raises Trellis.Error.
@spec dump_definitions_sql(trellis()) :: {:ok, String.t()} | {:error, Trellis.Error.t()}
The SQL that loads this instance's definitions into a database built from
a schema-only dump (pg_dump --schema-only, mix ecto.dump). Append it
to the dump; mix trellis.dump does.
It holds INSERTs for the rows that say what is defined (Trellis's
migration history, its instance marker, the transform definitions and the catalog rows they need), and nothing else: the ledger,
the ring, backfill progress, quarantine and claims describe rows a
schema-only dump doesn't carry. Every definition loads as "defined, not
built" (:waiting_to_backfill), the state a define/2 commits, including
one that was paused or quarantined where it was dumped. The rows are one
repeatable-read snapshot of the database. migrate/1 on the loaded
database does nothing.
Like dump_definitions_sql/1, but raises Trellis.Error.
@spec has_live_drain_workers(trellis()) :: {:ok, boolean()} | {:error, Trellis.Error.t()}
Whether at least one drain worker is alive anywhere in the fleet. With none, nothing reaches a target table: poll this from a health check.
Like has_live_drain_workers/1, but raises Trellis.Error.
@spec has_live_staging_worker(trellis()) :: {:ok, boolean()} | {:error, Trellis.Error.t()}
Whether the staging worker (which installs change capture and starts backfills) is alive anywhere in the fleet. The other half of the health check.
Like has_live_staging_worker/1, but raises Trellis.Error.
@spec migrate(trellis()) :: :ok | {:error, Trellis.Error.t()}
Creates or upgrades Trellis's own tables in the configured schema. Safe to run on every boot.
Run it on a handle with the default options, before connecting one that
runs background work: the staging worker reads Trellis's tables as it
starts, so a staging: true connect to an unmigrated schema fails with a
:not_found error.
@spec migrate!(trellis()) :: :ok
Like migrate/1, but raises Trellis.Error.
@spec poisoned_since(trellis(), DateTime.t()) :: {:ok, [Trellis.PoisonEntry.t()]} | {:error, Trellis.Error.t()}
Every source row the apply path poisoned (gave up on) after since,
oldest first. Poll it with the last entry's poisoned_at so a
whole-table failure doesn't sit unnoticed.
@spec poisoned_since!(trellis(), DateTime.t()) :: [Trellis.PoisonEntry.t()]
Like poisoned_since/2, but raises Trellis.Error.
@spec quarantine_status(trellis(), String.t()) :: {:ok, Trellis.QuarantineEntry.t()} | {:error, Trellis.Error.t()}
The state of one target: a transform ("order_totals") or one of its
columns ("order_totals.total"). A column that isn't paused is :live.
@spec quarantine_status!(trellis(), String.t()) :: Trellis.QuarantineEntry.t()
Like quarantine_status/2, but raises Trellis.Error.
@spec quarantined(trellis()) :: {:ok, [Trellis.QuarantineEntry.t()]} | {:error, Trellis.Error.t()}
Every quarantined transform and paused column, across every transform. Cheap enough for a dashboard or health check to poll.
@spec quarantined!(trellis()) :: [Trellis.QuarantineEntry.t()]
Like quarantined/1, but raises Trellis.Error.
@spec release_key(trellis(), String.t(), String.t(), String.t()) :: :ok | {:error, Trellis.Error.t()}
Releases one key transform holds in quarantine, once its cause is fixed.
source_table ("public.orders" or "orders") and key are as
sample_quarantined/3 and poisoned_since/2 report them.
It stages a recompute of the key, which every transform reading the table
applies from the key's current row, and discards the changes held for
transform meanwhile. Another transform holding the same key keeps
holding it. If the cause is still there, the key is poisoned again.
Resuming the transform releases every key it holds.
An unknown transform is a :not_found error, and so is a key it doesn't
hold, which changes nothing. The release first waits for the drain pages in
flight on the table to commit; a wait past the lock timeout (30 seconds) is
a :timeout error and changes nothing: call it again.
:ok = Trellis.release_key(trellis, "order_totals", "public.orders", "42")Like release_key/4, but raises Trellis.Error.
@spec request_backfill(trellis(), String.t()) :: :ok | {:error, Trellis.Error.t()}
Rebuilds every definition that reads source_table (a bare table name,
resolved like one in a statement), and returns once that rebuild is
queued.
Each :live reader reports :backfilling (rebuilt by the call itself)
until the re-read has re-derived every row the table still has and
deleted any target row it no longer backs. A newly defined transform
doesn't need this: its backfill is queued for it. Only a table Trellis
already captures can be re-read; any other is refused.
Like request_backfill/2, but raises Trellis.Error.
@spec sample_quarantined(trellis(), String.t(), [sample_option()]) :: {:ok, Trellis.SamplePage.t()} | {:error, Trellis.Error.t()}
One page of the rows quarantined under target, to diagnose a
quarantine's cause: for a column ("order_totals.total"), the rows that
failed evaluating it; for a whole transform ("order_totals"), the keys
poisoned from its source table.
{:ok, page} = Trellis.sample_quarantined(trellis, "order_totals.total", limit: 50)
{:ok, next} = Trellis.sample_quarantined(trellis, "order_totals.total", limit: 50, after: page.next_cursor)@spec sample_quarantined!(trellis(), String.t(), [sample_option()]) :: Trellis.SamplePage.t()
Like sample_quarantined/3, but raises Trellis.Error.
@spec self_check(trellis(), String.t(), [self_check_option()]) :: {:ok, Trellis.SelfCheckJob.t()} | {:error, Trellis.Error.t()}
Starts a background check of target_table against a fresh recompute of
its definition from the source tables, and returns the Trellis.SelfCheckJob
at once. See self_check_option/0.
The comparison is a drain worker's, a page of keys at a time (with no
drain worker anywhere in the fleet the job stays :queued, as a define
does). Poll self_check_job/2 with the job's id until it is finished
(Trellis.SelfCheckJob.finished?/1); its Trellis.SelfCheckReport is then
the verdict:
{:ok, job} = Trellis.self_check(trellis, "order_totals", timeout_ms: 30_000)
{:ok, %Trellis.SelfCheckJob{state: :done, report: report}} = Trellis.self_check_job(trellis, job.id)A target whose job is still :queued or :running gets that job back,
whatever options this call passed; a finished job stays until the next
self_check/3 of its target replaces it. Only a one-row-per-source-key
transform can be audited; an aggregate target is a :validation error,
and an unknown one
:not_found. A column that is paused is left out of the comparison.
The call reads nothing of the target, so it returns well inside the 30-second call limit.
@spec self_check!(trellis(), String.t(), [self_check_option()]) :: Trellis.SelfCheckJob.t()
Like self_check/3, but raises Trellis.Error.
@spec self_check_job(trellis(), integer()) :: {:ok, Trellis.SelfCheckJob.t() | nil} | {:error, Trellis.Error.t()}
The Trellis.SelfCheckJob that self_check/3 returned with this id, as
it stands now: its state, how many keys it has compared so far, and, once
it is :done, its report. nil when there is none: a newer self_check/3
of its target replaced it, or its transform was dropped.
@spec self_check_job!(trellis(), integer()) :: Trellis.SelfCheckJob.t() | nil
Like self_check_job/2, but raises Trellis.Error.
@spec shutdown(trellis()) :: :ok | {:error, Trellis.Error.t()}
Stops the handle's background work and waits for its threads to exit. Calls
still in flight are cancelled, not waited for: their callers get a
:validation error, like any later call on the handle. Shutting down again
is :ok.
@spec shutdown!(trellis()) :: :ok
Like shutdown/1, but raises Trellis.Error.
@spec start_link([start_option()]) :: GenServer.on_start()
Starts a process that owns a handle: it connects with options (see
start_option/0) as it starts, and shuts the handle down as its
supervisor stops it. Put it in a supervision tree as {Trellis, options}
rather than calling this directly.
Returns {:error, %Trellis.Error{}} if the connect fails, so the
supervisor's start fails with the reason. A second staging: true
process for the same instance fails that way (:conflict) while the first
is alive: in a rolling deploy, stop the old one before the new one starts.
The process only keeps the handle's lifecycle. A call made through its
name asks it for the handle and then runs in the caller's own process, so
callers don't wait on each other. The caller has to be on the process's
node: a handle works only on the node that connected it, and a call from
another node returns a :validation error.
shutdown/1 refuses the process's name: its supervisor stops it, which
cancels the calls still in flight (their callers get a :validation
error). The child spec gives it 30 seconds to shut down, since the
background threads have to exit; change it with Supervisor.child_spec/2.
A process killed before it shuts down leaves the handle to be shut down
when it is garbage collected.
@spec status(trellis(), String.t()) :: {:ok, Trellis.Status.t() | nil} | {:error, Trellis.Error.t()}
The status of the transform that writes target_table, or nil if none
does. target_table is either the fully qualified name define/2 and
apply/2 return ("public.order_totals") or the bare table name
("order_totals"). A qualified name with another schema matches nothing.
@spec status!(trellis(), String.t()) :: Trellis.Status.t() | nil
Like status/2, but raises Trellis.Error.
@spec watermark_token(trellis()) :: {:ok, watermark()} | {:error, Trellis.Error.t()}
A read-your-writes token covering every write committed before this call.
Take it after a source-table write commits, then pass it to
await_converged/3 to wait for that write to reach its targets.
Like watermark_token/1, but raises Trellis.Error.