# Trellis for Elixir Embedded Trellis for Elixir apps: a [Rustler](https://github.com/rusterlium/rustler) NIF over the `trellis` crate's `BlockingTrellis`, so a Phoenix app can define and run streaming transforms without deploying a separate service. The design is [ADR-0010](https://github.com/salesforce-misc/trellis/blob/main/docs/decisions/0010-embeddable-clients.md). ## Installation The Hex package is `trellis_pg` (Hex's `trellis` is taken); the modules are `Trellis.*`, and the OTP application is `:trellis_pg`. ```elixir def deps do [{:trellis_pg, "0.1.0-alpha"}] end ``` Installing needs no Rust: compiling the package downloads the NIF built for the machine from the release's GitHub page and checks it against the SHA-256 the package ships. Builds exist for - Linux, glibc 2.17 or newer: `x86_64` and `aarch64` - Linux, musl (Alpine): `x86_64` - macOS: Apple silicon and Intel On any other target, or to build it yourself, build the NIF from source. It needs a Rust toolchain (the version in the repository's `rust-toolchain.toml`) and Rustler in the host's own deps: ```elixir {:rustler, "~> 0.38.0", runtime: false} ``` ```sh TRELLIS_PG_BUILD=1 mix deps.compile trellis_pg --force ``` The choice is made when `trellis_pg` compiles, so `--force` it after changing the variable. `config :rustler_precompiled, :force_build, trellis_pg: true` does the same from config. Cargo fetches the `trellis` crate from this repository at the released commit. The NIF is built without the `trellis` crate's `otlp` feature, so neither path pulls in an OpenTelemetry or gRPC stack. The engine's log lines reach `Logger` through `Trellis.LogBridge`. ## Usage The surface mirrors the Rust crate's `BlockingTrellis`. Every function has a bang variant. - `connect/1`, `migrate/1`, `config/1`, `shutdown/1`: the handle's lifecycle. - `apply/2`: any statement of Trellis's grammar (`TRANSFORM`, `PAUSE TRANSFORM`, `RESUME TRANSFORM`, `DROP TRANSFORM`, `ALTER TRANSFORM`), returning what it did as a tagged value (`{:transform_defined, target_table}`, `:paused`, `{:resumed, addresses}`, ...). `define/2` is an `apply/2` that refuses anything but a `TRANSFORM`. - `status/2`, `definitions/1`: the catalog reads. - `quarantined/1`, `quarantine_status/2`, `sample_quarantined/3`, `poisoned_since/2`: the quarantine reads. Resuming is a statement. `release_key/4` releases one key a transform holds, once its cause is fixed; `status/2` reports how many it holds (`held_keys`). `status/2` also reports a page the drain keeps failing on with nothing charged or paused (`drain_failure`, a `Trellis.DrainFailure`), and a finished `self_check/3` job's report lists every one open (`drain_failures`). `status/2` names the paused upstream columns (`paused_upstream_columns`, each `"target.column"`) that make a `:live` transform report `:catching_up`. `status/2` lists the source indexes a `GROUP BY` or `JOIN` target needs and lacks (`source_index_gaps`, each a `Trellis.SourceIndexGap` with `table`, `columns`, `role` (`:group_by`, `:join_from` or `:join_joined`: the part of the definition that needs it), `opted_out` and a `create index` statement as `fix`); `self_check/3` refuses such a target unless it was defined `WITHOUT SOURCE INDEX`. `status/2` lists the partitions of a partitioned source a transform is being corrected for while it reads `:backfilling` for them (`correcting`, each a `Trellis.CorrectingNode` with `node`, `since` and `keys_remaining`). - `request_backfill/2`, `detect_capture_disruptions/1` (pauses now the readers of every table whose Trellis capture was changed outside Trellis, and returns the tables), `has_live_drain_workers/1`, `has_live_staging_worker/1`, `watermark_token/1`, `await_converged/3`. - `self_check/3` and `self_check_job/2`: check a target against a fresh recompute from its source, as a background job. `self_check/3` starts it and returns a `Trellis.SelfCheckJob` (`id`, `target`, `mode`, `state`, `rows_compared`, `report`, `error`) at once; a drain worker runs it, and `self_check_job/2` reads it back by `id` until `state` is `:done` (`:queued` and `:running` are in progress; `:failed` and `:cancelled` carry an `error`), when `report` is a `Trellis.SelfCheckReport` (`outcome`, `divergences`, `rows_compared`, `truncated`, ...). A target whose job is still unfinished gets that job back; the next call replaces a finished one, and `self_check_job/2` returns `nil` for the replaced job or one whose transform was dropped. With no drain worker anywhere in the fleet a job stays `:queued`. - `start_link/1`, as `{Trellis, options}` in a supervision tree: a process that owns the handle, so every function above also takes its name. - `Trellis.Migration`: `define`, `apply` and `status` in an Ecto migration. - `dump_definitions_sql/1`, `mix trellis.dump` and `Trellis.StructureDump`: the definitions as SQL appended to the structure file (see [Schema dumps](#schema-dumps)). - `Trellis.Metrics.render_prometheus/0`: the process-wide metrics registry as Prometheus text, for a `/metrics` route the host already serves (the binding opens no port). It needs no handle. Every series carries the `trellis_instance` label (`/`, as in the log lines), so handles in one VM stay apart. - `Trellis.LogBridge`: started with the `:trellis_pg` application, it forwards the engine's log lines to `Logger` (`domain: [:trellis]`, the Rust module as `:target` metadata). Every line names its instance in the text, as `trellis_instance=/`, so handles in one VM stay apart. Set `config :trellis_pg, log_level: :info` to choose the most verbose level forwarded (default: `Logger.level()`, read once at start), or `config :trellis_pg, log_bridge: false` to forward nothing. A `tracing` subscriber installed by another NIF never sees the engine's lines, since each native library links its own `tracing`; see `Trellis.LogBridge`. A line can be dropped if the bridge falls behind, and a warning says how many; logging never blocks or crashes the engine. ```elixir # A deploy's migration step: the defaults run nothing in the background. {:ok, migrator} = Trellis.connect(url: "postgres://localhost/app") # TLS options go in the URL (docs/embedding.md, "TLS to Postgres"): # url: "postgres://db.example.com/app?sslmode=verify-full&sslrootcert=/etc/ssl/ca.pem" :ok = Trellis.migrate(migrator) :ok = Trellis.shutdown(migrator) # The app, once migrated. The staging worker reads the catalog as it starts, # so a `staging: true` handle can't connect before `migrate/1` has run. {:ok, trellis} = Trellis.connect(url: "postgres://localhost/app", staging: true, drain_threads: 2) {:ok, _definition} = Trellis.define(trellis, "TRANSFORM widget_prices FROM widgets SELECT price AS price") {:ok, %Trellis.Status{status: :live}} = Trellis.status(trellis, "widget_prices") # A column that paused on bad data: find the rows, fix them, resume it. {:ok, [%Trellis.QuarantineEntry{target: "widget_prices.price"} | _]} = Trellis.quarantined(trellis) {:ok, %Trellis.SamplePage{samples: rows, next_cursor: cursor}} = Trellis.sample_quarantined(trellis, "widget_prices.price", limit: 50) {:ok, {:resumed, ["widget_prices.price"]}} = Trellis.apply(trellis, "RESUME TRANSFORM widget_prices.price") :ok = Trellis.shutdown(trellis) ``` `connect/1`'s defaults (`staging: false, drain_threads: 0`) run nothing in the background. For each instance (catalog schema), exactly one connection in a fleet should set `staging: true`, and some connection must run drain threads, or no definition ever reaches `:live`. ### Conventions - Non-bang functions return `{:ok, value}` (or `:ok`) or `{:error, %Trellis.Error{code: code, message: message}}`; bang variants return the value or raise that error. `code` is one of a closed set of atoms, with `:unknown` for a code newer than the binding. - No atom is ever built from a string the database returned: every atom in a result comes from a closed set the NIF allocates when it loads. - Times are `DateTime`s. They cross the NIF as epoch microseconds. - A quarantine target is an address string, `"transform"` or `"transform.column"`, exactly as `quarantined/1` reports it. - `sample_quarantined/3`'s cursor and `watermark_token/1`'s token are opaque: pass back what the previous call returned. - A call on a handle holds no scheduler while it waits. The NIF submits the call on a normal scheduler and returns a reference at once; the engine sends the result to the calling process, which waits for it in a `receive`. The function is synchronous to its caller, a process killed during a call exits at once, and a call stuck on the database holds no scheduler or thread of its own. `connect/1` and `shutdown/1` run on dirty IO schedulers, as do the NIFs that touch no handle; `Trellis.Metrics.render_prometheus/0` runs on a dirty CPU scheduler. - Every call returns within 30 seconds, counted from when it is submitted (a database that can't answer in time gives `{:error, %Trellis.Error{code: :timeout}}`; the server stops what the call started). 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. - `shutdown/1` doesn't wait for calls in flight: it cancels them, and each of their callers gets a `:validation` error. ## In a Phoenix app Let the application's supervisor own the handle. `{Trellis, options}` starts a process that connects with `connect/1`'s options as it starts, hands the handle to the callers that make calls through its name, and shuts the handle down when the supervisor stops it: ```elixir # config/runtime.exs database_url = System.fetch_env!("DATABASE_URL") config :my_app, MyApp.Trellis, name: MyApp.Trellis, url: database_url, # Only the fleet's one worker node runs the background work. staging: System.get_env("TRELLIS_WORKER") == "true", drain_threads: if(System.get_env("TRELLIS_WORKER") == "true", do: 2, else: 0) # lib/my_app/application.ex children = [ MyApp.Repo, {Trellis, Application.fetch_env!(:my_app, MyApp.Trellis)}, MyAppWeb.Endpoint ] # Anywhere in the app: the name stands in for the handle. {:ok, status} = Trellis.status(MyApp.Trellis, "widget_prices") ``` Each call runs in the process that makes it, in parallel with every other, so no process queues behind another's call. The caller has to be on the same node: a handle works only on the node that connected it. `shutdown/1` refuses the name; the supervisor stops it, which cancels the calls still in flight. Define transforms in Ecto migrations with `Trellis.Migration`. Trellis never joins a migration's transaction, so a migration that uses it must set `@disable_ddl_transaction true` (the helpers raise before applying anything if a transaction is open), and must define `up/0` and `down/0`: ```elixir defmodule MyApp.Repo.Migrations.DefineWidgetPrices do use Ecto.Migration use Trellis.Migration @disable_ddl_transaction true def up do define "TRANSFORM widget_prices FROM widgets SELECT price AS price" end def down do apply "PAUSE TRANSFORM widget_prices" apply "DROP TRANSFORM widget_prices" end end ``` Ecto doesn't start the application to migrate, so each helper connects a handle of its own, from the repo's `:trellis` configuration (`config :my_app, MyApp.Repo, trellis: [url: database_url]`), and runs `migrate/1` on it first. A define isn't idempotent, and the helpers don't make it so: Ecto's `schema_migrations` runs each migration once. See `Trellis.Migration` for the details, and for an optional guard. Add `import_deps: [:trellis_pg]` to the app's `.formatter.exs` to keep `define` and `apply` free of parentheses, like Ecto's own commands. ## Fixtures and bulk loads in tests A loader that disables triggers for the load (`ALTER TABLE ... DISABLE TRIGGER ALL`, `pg_restore --disable-triggers`) disables Trellis's capture triggers too, and `ENABLE TRIGGER ALL` leaves them in a mode Trellis treats as a disruption. The rows it wrote never reached a target, so Trellis pauses every definition that reads the table (`capture_failure` has `kind: :disruption`) instead of letting a test read silently wrong data, and waits for a `RESUME`. `Trellis.TestSupport` does that `RESUME` for a suite, then waits for the rebuilt targets to catch up: ```elixir setup do # ... load the fixtures with triggers disabled, then re-enabled ... {:ok, _resumed} = Trellis.TestSupport.resume_capture_disruptions(trellis) :ok end ``` `resume_capture_disruptions/2` first pauses the readers of any disrupted table itself (`Trellis.detect_capture_disruptions/1`), so it doesn't wait for the staging worker's next pass; resumes only the definitions paused with kind `:disruption`; waits until each is `:live` and every write committed so far has reached its targets; and returns `{:ok, resumed}` (`[]` when nothing was disrupted, which costs one catalog read per captured table). A transform an operator paused, or one paused by a schema change or another failure, stays paused. A `RESUME` the engine refuses (a table that is gone, a table define would now refuse) leaves that definition paused; the rest are still resumed, and the call then returns `{:error, {:resume_refused, refused}}`, a map of target to reason. `resume_capture_disruptions!/2` returns the targets and raises `Trellis.TestSupport.ResumeRefused` (or `Trellis.Error`). The `:timeout_ms` option (default 30 000) bounds the wait. The handle needs its staging worker and a drain worker (`staging: true, drain_threads: 1`), because the resume rebuilds through them. Trellis sees a write when it commits, so with Ecto's SQL sandbox, load the fixtures that targets must reflect outside the sandbox transaction (in `setup_all`, or with the sandbox in `:auto` mode) and call the helper once. ## Schema dumps `mix ecto.dump` writes a schema-only `pg_dump`, which carries the Trellis schema, the target tables and the capture triggers but no rows, so a database `mix ecto.load` builds from it has targets and no transforms. `mix trellis.dump` runs `mix ecto.dump` and appends the definitions the way Ecto appends `schema_migrations`. Make it what `mix ecto.dump` runs, by an alias in `mix.exs`: ```elixir defp aliases do ["ecto.dump": "trellis.dump"] end ``` The task passes its arguments (`-r`, `-d`, `--prefix`, `-q`) to `ecto.dump` unchanged. It reads the definitions before and after the `pg_dump` and redoes the dump, up to three times, when a define or drop committed in between, so the file's DDL and definitions agree. The repo needs its `:trellis` configuration (as for `Trellis.Migration`); the definitions are appended for each repo that has one. If the dump names its schemas (`--prefix`, or the repo's `:dump_prefixes`), the list has to include the Trellis schema, and the task refuses one that doesn't. `mix ecto.load` then runs the file and nothing else: the worker that starts on the loaded database finds definitions that aren't built and builds them. Only definitions are dumped; the ledger, backfill progress, quarantine and claims are runtime state, and a paused or quarantined transform loads `:waiting_to_backfill`. `pg_dump` 17 writes `CREATE SCHEMA public;` when it dumps `public` by name, which stops `mix ecto.load` into a database `mix ecto.create` made; the task writes `CREATE SCHEMA IF NOT EXISTS public;` instead. `Trellis.StructureDump.append/2` appends to a structure file someone else dumped (reading the definitions once, so it can't see a define racing that dump), and `Trellis.dump_definitions_sql/1` returns the SQL for a caller that assembles its own dump. ## Layout - `lib/`: the public `Trellis` module and its structs, the process that owns a supervised handle (`lib/trellis/owner.ex`), and `Trellis.Migration`, which is compiled only when the host depends on `ecto_sql`. - `lib/trellis/test_support.ex`: `Trellis.TestSupport`, the test helper above. - `native/trellis_nif/`: the NIF crate, a member of the repository's Cargo workspace. The release makes the packaged copy stand alone (see Releasing). Plain-data conversion and error codes come from `clients/embed` (`trellis-embed`), shared with the Ruby binding. ## Development The Erlang and Elixir versions are pinned in `.tool-versions`. From this directory: ```sh cargo build -p testkit --bin trellis-testkit # the suite's throwaway Postgres mix deps.get mix test ``` In this repository (`:dev` and `:test`) the NIF always builds from source, never downloads. `mix test` compiles it through Cargo into the workspace's `target/`, and the suite spawns `trellis-testkit` (from `PATH`, else the workspace's debug build) for a Postgres cluster that is torn down when the test run exits. It needs the Postgres server binaries on `PATH`, like the Rust tests. `test/trellis/parity_test.exs` runs `clients/parity/cases.json`, the fixture the Ruby suite runs too, so the two bindings can't drift apart unnoticed (see `clients/parity/README.md`). ## Releasing `trellis_pg` is released with the rest of Trellis, from one `v` tag ([docs/releasing.md](../../docs/releasing.md)). `@version` in `mix.exs` and the NIF crate's version (`native/trellis_nif/Cargo.toml`) are the release's. `.github/workflows/release.yml` builds the NIF for each target above and attaches them to the tag's GitHub release, where `lib/trellis/native.ex` downloads them from. The package is then published to Hex by hand with `.github/scripts/publish-packages.sh`, which writes `checksum-Elixir.Trellis.Native.exs` from the NIFs that release serves. `.github/scripts/build-elixir-nif.sh ` builds one NIF the way the workflow does. Linux targets need `cargo-zigbuild` and Zig (`pip install cargo-zigbuild ziglang`).