Data Pipelines

Lana's reporting pipeline runs inside the application. The operational database publishes public events, the outbox relay lands those events as append-only Parquet batches in a local landing zone, dbt transforms them on DuckDB, and the Rust reporting jobs render downloadable report files from a published serving copy.

Overview

The pipeline provides:

• Event-sourced warehouse tables for reporting and analytics.
• dbt transformations in dw, running on DuckDB.
• Regulatory and operational report files generated by Rust jobs in core/report.
• CI checks that compare public event schemas, dbt demux models, and report definitions.

DuckDB Warehouse

The data warehouse is DuckDB — there is no warehouse database server. The application needs a single local directory, configured as dw_db.path (environment variable LANA__DW_DB__PATH). In Kubernetes, the deployment mounts the lana-dw PVC at /var/lib/lana/dw; local dev stacks point it at .nix-deps/dw in the worktree. An empty path leaves the warehouse unconfigured: the relay lands nothing, no builds run, and manual and end-of-day report generation are disabled for that deployment.

The directory holds all of the pipeline's local state:

<dw root>/landing/events/events-<first>-<last>.parquet        relayed public events
<dw root>/landing/party_pii/party-pii-<first>-<last>.parquet  relayed party-PII stream
<dw root>/landing/_checkpoint.json                             committed event prefix
<dw root>/build/lana_dw.duckdb                                 dbt build database
<dw root>/serve/lana_dw.duckdb                                 published serving copy

Landing Zone

The outbox relay (inside the lana app) lands public events and the party-PII stream as append-only Parquet batches: one file per committed flush batch per stream, named by the commit-sequence range it covers. After each batch, the relay atomically renames _checkpoint.json, which names the committed event prefix — every event with a commit sequence up to that value is durably present in the landing zone. A crash between a batch write and its checkpoint leaves orphan Parquet files beyond the checkpoint; the next flush deletes those orphans before writing, so replay never duplicates rows.

By default the landing zone is a local directory under the DW root. Production deployments configure an object-storage mirror via the dw_db.landing config section:

dw_db:
  path: /var/lib/lana/dw
  landing:
    provider: gcp
    bucket: galoystaging-lana-bank-<instance>-dw
    prefix: dw

The local directory stays the working copy — the relay and the dbt build only ever touch it — while every batch and checkpoint is also uploaded to the bucket through the pod's workload identity (no credentials configured). On pod start, an empty local landing is restored from the mirror before the relay runs, so the DW root can be ephemeral (emptyDir): the deployment needs no PVC, and the build/serve databases are rebuilt by the initial pipeline run after the restore. Flush order keeps the mirror authoritative: local batch files → object uploads → object checkpoint → local checkpoint, so the bucket's checkpoint never names data it lacks and a crash anywhere is resolved by the next replay.

Rollout note: enabling the mirror on an instance that already has a landed history bootstraps the bucket from the local landing on first boot — but that local history must still exist at flip time. Migrate an existing PVC-backed landing (upload its contents to the bucket) before removing the volume.

dbt Build

dbt runs on dbt-duckdb only (profile dbt_lana_duckdb_dw in dw/profiles.yml). Sources read the landing zone via external locations, and the build writes <dw root>/build/lana_dw.duckdb; incremental state persists across builds. The profile takes both locations from the environment: LANA_DUCKDB_PATH (the build database) and LANA_DW_LANDING_DIR (the landing zone).

The build runs as its own process in the application pod, and its whole DuckDB engine — one engine for the run, however many models dbt executes concurrently — is bounded through the profile's settings: block, configurable per deployment via the dw_db.build section (environment variable prefix LANA__DW_DB__BUILD__, e.g. LANA__DW_DB__BUILD__MEMORY_LIMIT):

dw_db:
  build:
    memory_limit: 3GB            # engine memory limit for the whole build
    threads: 2                    # engine worker threads
    max_temp_directory_size: 20GB # spill-directory cap (dw-data volume)

DuckDB's defaults size the engine to the node — a worker thread per node core and memory_limit at 80% of node RAM — so an unbounded build can spend the pod's entire CPU and memory budget while GraphQL serving shares it; a full-refresh build that froze all inbound serving for its duration is what prompted these bounds (mirroring the serving-side bounds of dw_db.serve). The defaults above trade build speed for not starving co-located serving and fit the 3-CPU/6Gi pod class with a 40Gi dw-data volume: work that no longer fits in memory_limit spills to the temp directory, and a build that cannot fit either fails as a rejected build rather than taking the pod down. Every field is optional: a partial section overrides only what it names, and the values shown are the defaults. dbt's own model concurrency (threads: 2 in the profile) is a separate knob and stays as is. Direct dbt invocations (BATS, CI) inherit the same defaults because the profile carries them as fallbacks; the app delivers its configured values as the LANA_DW_DBT_ENGINE_* environment variables.

Serving Copy

After every successful full or intraday build, the application publishes an atomic serving copy at <dw root>/serve/lana_dw.duckdb: it checkpoints the build database's WAL, copies the file next to the serving path, and renames it into place. The copy keeps the same file name as the build database so the catalog-qualified references dbt bakes into stored views and report macros keep resolving. Report jobs and the read-only query endpoint open the serving copy read-only and never contend with a build; new readers pick up a freshly published copy on their next connection.

Readers go through a bounded connection pool whose budget is configurable per deployment via the dw_db.serve section (environment variable prefix LANA__DW_DB__SERVE__, e.g. LANA__DW_DB__SERVE__MEMORY_LIMIT):

dw_db:
  serve:
    memory_limit: 256MB           # per-connection engine memory limit
    threads: 1                    # worker threads per connection
    max_temp_directory_size: 1GB  # per-connection spill-directory cap
    max_connections: 4            # live connections; further queries queue
    interrupt_grace_secs: 30      # deadline grace before a statement is declared abandoned

Each pooled connection is its own embedded DuckDB instance, so the worst-case serving footprint is max_connections × memory_limit — sized to the pod, not the node. Every field is optional: a partial section overrides only what it names, and the values shown are the defaults. A query that cannot fit its execution inside memory_limit comes back as a rejected query, and a query that cannot acquire a connection in time fails the same way a slow statement would.

Statements run under the caller's deadline (15s on the read-only query endpoint). When the deadline fires, the statement receives repeated interrupts and its connection slot stays held — pairing the engine memory it still occupies — until the statement actually stops, so timed-out statements cannot multiply live engines past max_connections. A statement that keeps ignoring interrupts for interrupt_grace_secs is declared abandoned: reported once at ERROR level with a statement fingerprint, then interrupted on a slow background cadence. One thread per instance keeps an abandoned statement's CPU footprint to a single core, bounding the whole pool's grind to max_connections cores. A statement stuck in a non-interruptible native allocation cannot be preempted in-process; the SQL validation boundary is the front line for those shapes, and serving isolation is the durable fix.

dbt Transformations

The dbt project lives at:

dw/
├── models/
│   ├── staging/
│   │   └── lana_events/
│   │       └── demux/
│   ├── intermediate/
│   │   ├── streams/
│   │   ├── reference/
│   │   ├── enriched/
│   │   ├── entities/
│   │   └── core/
│   └── outputs/
│       ├── marts/
│       └── reports/
├── macros/
├── seeds/
├── dbt_project.yml
└── profiles.yml

Seed filenames use the static_ prefix followed by a descriptive snake_case name.

The model tiers are:

Tier	Purpose
Staging	Source-conformed inputs. Public events are split into generated typed variants under staging/demux/ (stg_<module>_<variant>); only intermediate streams may read those variants. PII is staged separately as stg_party_pii.
Intermediate	Purpose-built transformations. Its streams/ models form append-only entity logs (int_<entity>_events); reference/, enriched/, and entities/ retain their specialized roles. Shared interval, balance, and set-algebra business logic lives in core/; report-scoped regulatory logic lives in the norm folders. No model sits directly at the intermediate/ root — every model lives in a named subfolder.
Outputs	Terminal interfaces for downstream consumers. marts/ contains conformed fct_* / dim_* relations for analytics; reports/ contains final report relations and report macros consumed by the Rust report runner. Each retains its own contract and schema.

Ledger account, account-set membership, transaction, entry, and effective-balance events publish GL-qualified payload fields directly, such as gl_account_id, gl_transaction_effective_date, and gl_entry_amount_native. Generated staging and stream models preserve those names, so accounting marts do not need to repeat naming-only projections. Transaction-template events keep their template-oriented field names.

As-of reports use the as_of_sql_function materialization, which builds a DuckDB table macro — report_x(date) — callable from SQL exactly like the previous Postgres SQL functions, with the as-of date bound by the caller.

PII Handling

The party_pii landing stream is append-only and carries erasure tombstones: a row with a NULL payload marks the party's PII as erased. The current-PII snapshot is derived in stg_party_pii — the latest commit_sequence per party wins, and a tombstone means no PII surfaces downstream. That snapshot stays a non-persisting view, per the dw lint's PII guardrails.

Refresh Triggers

Warehouse refreshes are tracked under Pipeline Runs in the admin panel:

Kind	Trigger	Scope
Initialization	Lifecycle.InitializationCompleted, once per banking database	Full dbt build, including seeds and tests
End of day	End-of-day processing completes	Full dbt build, including seeds and tests
Intraday	Hourly clock event, after the first successful full build	Operational marts and their seeds (dbt build, without tests)
On demand	Run now in Pipeline Runs, or the admin API	Full dbt build by default; the API also supports partial operational-mart refreshes

All kinds share one queue, so dbt never runs concurrently. Full builds wait for the warehouse outbox relay to catch up before starting.

After all synchronous core banking startup initialization succeeds, the application publishes Lifecycle.InitializationCompleted. This durable milestone is emitted once per database, including when replicas start concurrently; failed initialization does not emit it, and restarts do not repeat it. An existing database without the milestone emits it on its next successful startup. It does not mean that asynchronous integrations or operator-provided product/accounting configuration have completed.

When a warehouse is configured, the event queues an Initialization run to create the warehouse models and populate seeds and available startup data without waiting for EOD or an hourly tick. Business tables with no activity may still have no rows, and reporting bounds remain empty until the first EOD completes. A chart is not required for the initial build; once configured, the normal chart-mapping validations apply. The consumer replays the durable event if the warehouse is enabled later. Completion enables hourly refreshes but does not certify a settled EOD reporting day. Failed builds use the normal pipeline retry policy; after retries are exhausted, operators can request an on-demand full build.

When scheduled refreshes arrive faster than the warehouse can process them (for example, during accelerated simulated-time demos), pending work is compacted. Only the latest pending end-of-day run is kept; if no end-of-day run is pending, only the latest hourly run is kept. End-of-day builds take priority over hourly refreshes because they rebuild and test the full DAG. Superseded runs appear as Skipped, not completed, and do not advance report freshness. Skipping warehouse refreshes does not skip the bank's end-of-day accounting processing.

A running build is never interrupted, and ticks arriving during it can queue a follow-up refresh. Retries, initialization builds, and explicit on-demand requests are not compacted. This catch-up policy applies to scheduled backlogs in both real and simulated time.

On-demand refreshes require the dw_pipeline_writer permission set, granted to admins and bank managers by default. They can run without waiting for an hourly event or advancing the operational clock. Their day is the request's operational UTC date, not a settled reporting day: completing one does not advance end-of-day report freshness or replace EOD processing.

Use refreshMode: FULL (the default) to build the entire DAG, including seeds and tests, or refreshMode: PARTIAL to build the intraday operational-mart selector and its seeds without tests. Scheduled hourly refreshes begin only after a full build has completed. Like hourly refreshes, on-demand partial refreshes proceed best-effort if the relay does not catch up within the partial-refresh timeout. The admin panel's Run now action uses FULL.

To request a partial refresh through the admin GraphQL API:

mutation {
  pipelineRunRequest(refreshMode: PARTIAL) {
    ... on PipelineRunRequestPayloadSuccess {
      pipelineRun {
        pipelineRunId
        kind
        refreshMode
        status
      }
    }
    ... on PipelineRunRequestPayloadError {
      code
    }
  }
}

The mutation returns the queued run immediately. Follow its status and per-node results in the admin panel or query pipelineRun(id: ...). Each request creates a separate run whose refreshMode is retained across retries. If the warehouse is not configured, the mutation returns WAREHOUSE_NOT_CONFIGURED without queuing a build.

Report Generation

Report downloads query the published serving copy. The report runner binds dates and reads the serving copy read-only through the embedded DuckDB engine (the Rust duckdb crate), renders CSV, XML, or text using the definitions in core/report/report-definitions/reports.yml, and uploads the result to configured storage. Downloading a report does not itself trigger dbt; warehouse refreshes use the triggers above.

The read-only ad-hoc query endpoint validates submitted SQL with sqlparser, then runs it read-only against the serving copy with search_path mart,main and a row cap. There is no statement timeout — the row cap is the limit.

CI Checks

The data pipeline checks validate the dbt-duckdb project without provisioning warehouse infrastructure: dbt parse only renders the project, so scratch DuckDB locations are enough.

• dev/check-dw.py uses the dbt manifest to validate model dependency boundaries, direct stream watermark dependencies, report and mart conventions, and PII materialization.
• dev/generate-dw-demux.py check verifies that generated public-event staging models match the schema and the bundle manifest (dw/stream-events.yml).
• dev/generate-dw-streams.py check verifies that generated event-stream models match the schema and bundle manifest.
• int_gl_transaction_template_entries carries the reporting classification of each template entry (declared as cala entry metadata at template definition time and mirrored atomically onto Accounting.LedgerTxTemplateCreated; cala templates are create-once, so publish-once is complete), so a classification cannot desync between code and the warehouse. int_cala_classified_transaction_amounts derives its mapping from that stream, and a dbt_utils.unique_combination_of_columns test on the classified combinations guards the join against double-counting.
• ci/tasks/validate-report-columns.sh parses the dbt manifest and checks that as-of report definitions point at as_of_sql_function models using the expected as-of filter macro.