Sources

Sources are the extract/load side of Frank. A source connects to one system, discovers its streams, and lands selected data in tenant-scoped Bronze Iceberg tables.

Source lifecycle

draft -> ready -> syncing -> active <-> paused
                         |
                         +-> error
* -> decommissioned

This lifecycle only describes EL. It does not say whether a transform exists or whether downstream data is modeled.

Source patterns

A source starts from a pattern in backend/config/patterns. Patterns define:

• Display metadata: id, name, description, category, complexity, icon.
• Engine: airbyte or dlt.
• Connector config: Airbyte source image or dlt source type.
• Field definitions: required and optional form fields.
• Defaults and templates.
• Examples, supported formats, auth methods, and transformation hints.

Current pattern coverage includes:

Category	Examples	Engine mix
Databases	PostgreSQL, MySQL, SQL Server, MongoDB	Airbyte
Warehouses	BigQuery, Snowflake, Redshift, Databricks	Airbyte
CRM and finance	Salesforce, HubSpot, Stripe	Airbyte and dlt
APIs	REST, GraphQL, GitHub, Jira, Notion, Slack, Airtable, RSS	dlt and Airbyte
Files	S3, SFTP bulk, Google Sheets, filesystem, archive/ZIP	Airbyte and dlt
Streams	Kafka	dlt

Patterns are synced to the database at API startup, so the UI can render dynamic forms without hardcoded connector fields.

Choosing an engine

Use case	Recommended engine	Why
Mature SaaS or database connector	Airbyte	Existing connector behavior, schema discovery, and Docker isolation.
Custom REST or GraphQL API	dlt	Python-native source construction, pagination/auth templates, nested JSON normalization.
Filesystem or lightweight custom extraction	dlt	In-process readers and multi-table normalization.
High-volume production replication	Airbyte	Better fit for source-defined sync behavior and connector ecosystem.

Frank hides most engine differences behind the ExtractionEngine interface. Both engines produce data envelopes, cursor state, batch progress, and Iceberg writes.

Credentials and runtime modes

Pattern fields marked credential: true or type: password are structurally credential-bearing. Their values are write-only: Source list/detail, dry-run, pipeline export, logs, traces, and errors never return them. The deployment selects one exact SOURCE_CREDENTIAL_MODE; Frank does not infer it from Vault reachability, environment variables, or existing rows.

Current base Compose: legacy_inline

Base docker-compose.yml uses legacy_inline so the mutable developer stack remains runnable without Vault during Generation A. Source create/update accepts the product's write-only credential_values and merges them into Source.source_config at the persistence boundary. Ordinary configuration updates preserve hidden values that redacted readback cannot return. Discovery and extraction use the inline configuration while results and failures remain redacted.

A repeatable Git-owned manifest still contains only ordinary configuration. Provide credential values separately through the Source UI or a transient write-only request that is never committed or placed in shell arguments. pipelines export strips registered credential forms. Vault references and credential lifecycle operations are unavailable in legacy_inline; the runtime also refuses readiness if the database already contains a credential_ref.

Immutable release: vault

The immutable release overlay uses vault. The API and workers refuse readiness when their service-specific Vault identities cannot perform exactly their reviewed capabilities.

Frank rejects declared credential keys in ordinary source_config; values are accepted only through write-only API/UI inputs, stored in tenant-isolated Vault KV v2, and represented in PostgreSQL by an opaque credential_ref plus safe metadata such as declared fields, exact version, presence, lifecycle state, and rotation time.

A repeatable Vault-mode Source keeps only ordinary configuration and the opaque reference:

apiVersion: frank.platform/v1
kind: Source
metadata:
  name: incoming_files
spec:
  pattern_id: s3
  credential_ref: 72c4d2ce-62aa-4f82-b8a3-a96881b2fd01
  source_config:
    bucket: city-data
    path_prefix: incoming/
    file_format: parquet
    region_name: eu-west-1

Bootstrap or rotate values non-interactively with the Vault-only commands:

frankctl sources credentials set <source-id> \
  --name incoming-files --values-file - < credential-values.yaml
frankctl sources credentials rotate <source-id> \
  --values-file - < rotated-values.yaml

The values object must exactly match the pattern's declared credential fields. Use stdin from the approved secret manager. If a temporary file is unavoidable, keep it untracked with mode 0600 and remove it through the approved secure procedure after use. The UI similarly sends a newly entered value once; it never loads the current value back into the browser.

Discovery and extraction carry only the reference and exact version through Temporal, then resolve that version inside the activity process immediately before connector execution. Rotation preserves the Source, streams, schedules, cursors, loaded-file ledger, and run history.

Before a deployment enters vault, the release cutover writes and verifies every registered value in Vault before attaching references and removing only declared fields from PostgreSQL in one transaction. Once that transaction commits, returning that database to legacy_inline or an older binary is not a supported recovery path. Recovery uses a compatible matched PostgreSQL/Vault artifact pair or a forward fix. Source deletion or detachment never deletes Vault data, and revocation is rejected while any Source still references the credential.

Streams

Each source owns streams. A stream is configured independently:

Field	Meaning
name	Source-side stream/table/resource name from discovery.
namespace	Optional source namespace or schema.
dest_table_name	Optional Bronze table override. Child tables inherit the override prefix.
sync_mode	full_refresh or incremental.
cursor_field	Cursor column/resource field for incremental syncs.
write_disposition	append, replace, or merge.
primary_key_path	Primary key fields for merge/upsert behavior.
is_enabled	Whether this stream participates in sync runs.
schema	JSON schema saved from discovery.

For incremental streams, Frank persists cursor state and uses overlap windows to avoid missing late-arriving rows.

Declarative stream ownership

A kind: Source document may declare spec.streams. On both create and reapply, Frank matches those streams by name, creates missing declarations, and patches changed configuration in place. Reconciliation never resets the stream ID, cursor, iterator chunk state, row counters, last-sync metadata, or Source run history. For existing streams, omitted configuration fields remain unchanged; API defaults apply only when a missing stream is created.

Omission is deliberately non-destructive. An existing stream absent from the manifest remains unchanged and server dry-run reports it as unmanaged with a warning. Declare is_enabled: false to disable a stream explicitly; removing a stream requires a separate explicit operation. An exact reapply performs no database write.

For pipelines apply --wait, a Source document with spec.streams syncs only the explicitly enabled streams in that declaration. This lets a replacement logical stream run without executing preserved unmanaged legacy streams. If the Source document omits streams, --wait retains its legacy behavior and syncs every persisted enabled stream.

Set spec.max_items_per_chunk to an integer from 1 through 2147483647 when a Source needs an explicit iterator chunk budget, for example max_items_per_chunk: 10000. Create, PATCH, idempotent reapply, server dry-run, and pipelines export preserve this Source-level value. Omitting it uses the connector or cluster default and does not overwrite an existing declaration during reapply. An explicit null in a PATCH clears an existing override so the Source returns to the connector or cluster default.

streams:
  - name: events
    sync_mode: full_refresh
    write_disposition: replace
    primary_key_path: [id]
    is_enabled: true
    dest_table_name: events
  # Migration declaration: disable an adopted legacy row without deleting it.
  - name: events_page_1
    is_enabled: false

For an adopted Source, explicitly list each known obsolete enabled stream as is_enabled: false. That makes scheduled and direct Source syncs converge too; relying only on --wait stream selection would leave those other execution paths able to run the legacy rows. The disable is an in-place config patch and does not erase the stream's runtime state.

frankctl pipelines export includes this config-only declaration. It excludes stream IDs, cursors, chunk state, counters, timestamps, dataset IDs, and schema learned at runtime.

Refreshing discovered schema without resetting routing

Refresh a Source directly from its connector discovery, or submit a previously captured complete discovery result:

frankctl sources streams refresh-schema <source-id> --from-discovery \
  --timeout 300 --poll-interval 2000
frankctl sources streams refresh-schema <source-id> -f discovery-result.yaml

For an existing stream matched by name, schema refresh changes only the saved schema (from discovery's json_schema) and supported_sync_modes. It preserves:

• operational routing: namespace and dest_table_name;
• operator choices: sync_mode, write_disposition, cursor_field, primary_key_path, and is_enabled; and
• runtime state, including stream ID, last cursor, iterator chunk state, last_sync_at, and rows_synced.

Schema refresh is therefore not a route migration. If a namespace, destination table, cursor, or write policy must change, make that an explicit stream configuration update.

In connector discovery output, namespace means a real upstream namespace that contains the stream, such as a database schema or GeoServer workspace. It is not the stream name, Frank's tenant-scoped Bronze namespace, or a destination table override. A flat REST/SOAP source with no upstream namespace should return null; returning namespace: towers for a towers stream can produce a doubled destination identity. Existing namespaces are preserved by refresh; newly discovered streams receive the connector-supplied source namespace.

Submit a complete discovery snapshot

Streams absent from a non-empty refresh result are removed as stale, while new names are created with defaults. A hand-authored -f payload must therefore contain the complete current stream list, not only the stream whose schema changed.

full_refresh must use replace, not append

A full_refresh stream re-fetches the whole snapshot each run. Pair it with write_disposition: replace — with the default append, every scheduled sync re-appends the entire dataset and Bronze balloons (a */5 full_refresh source was observed at ~200× duplication). Use append/merge only for incremental.

Verified-empty publication and row counters

Zero rows are publishable only when the connector fully exhausted the source successfully and produced connector-owned verified-empty evidence. A first verified-empty sync creates the declared typed Bronze table without a sentinel row. For an existing table, zero rows may replace its contents only when the stream is exactly full_refresh plus replace:

• iterator mode writes the empty snapshot on the run's WAP branch and advances main once after complete provenance is verified;
• monolithic mode commits one atomic empty overwrite directly to main; and
• both paths retain the existing schema and recognize retries after the write or publication boundary.

Unknown, partial, failed, cancelled, append, merge, and incremental empty results cannot erase an existing table. Frank does not drop or recreate the table and never fabricates a row to make an empty dataset visible.

Stream.rows_synced describes the current published table cardinality after a successful full_refresh + replace; repeating a 10,000-row replacement still reports 10000, and a verified-empty replacement reports 0. Append and incremental writes retain cumulative counter behavior. Failed or partial runs do not apply a terminal-success counter update.

One logical stream for a paginated REST dataset

Page-number APIs should be represented as one REST resource and one Stream, not as items_page_1, items_page_2, and so on. For an API that exposes a record array but no total/next metadata, make empty-page termination explicit:

source_config:
  base_url: https://api.example.test/v1
  resources: [/events]
  data_selector: data
  paginator: page_number
  page_param: page
  base_page: 1
  pagination_total_path: null
  stop_after_empty_page: true
  primary_keys: [id]

In the Source create or edit form, choose page_number from Response → Pagination. The schema-driven form then reveals Page Query Parameter, First Page Number, Total Pages Path, and Stop After Empty Page. These controls write the same configuration shown above; they are not a separate dashboard-only pagination mode.

The connector requests page 1 onward and stops at the first empty selected data array. Do not set maximum_page for a complete dataset: an artificial cap makes the manifest silently incomplete when the upstream grows. TLS verification remains enabled; certificate failures are terminal rather than an excuse to bypass verification.

Deterministic GeoServer WFS pagination

A geoserver_wfs Stream must declare its exact source-side primary key. Frank uses that ordered key as WFS sortBy; an explicit sort_by is accepted only when it names the same fields in the same order. This deterministic iterator currently supports full_refresh only; incremental mode fails explicitly rather than publishing a cursor from primary-key-ordered pages. Composite keys are supported:

source_config:
  base_url: https://geo.example.test/geoserver
  workspace: transit
  resources: [ParagensTotalSchedule]
  page_size: 10000
  sort_by: trip_id A,Ordem A,IdParagem A
streams:
  - name: ParagensTotalSchedule
    sync_mode: full_refresh
    write_disposition: replace
    primary_key_path: [trip_id, Ordem, IdParagem]
    is_enabled: true

The connector pins the query and numberMatched/totalFeatures, then records the page start and intra-page offset. It fails instead of continuing if the total, query, page prefix, feature identity, or page seam changes. Frank treats the GeoServer database's sortBy order as authoritative because its string collation may differ from Python's. Before every page after the first, Frank requests the preceding and next rows as a two-row seam: the preceding row must match the persisted boundary, and the next row must match the normal page's first row. All responses require unique feature IDs and key tuples plus consistent key types. Resuming within a page requires the same consumed prefix. Changing filters, sort fields, page shape, pagination strategy, or geometry/index output during an in-flight run therefore requires restarting that full refresh from its initial state.

For an existing Bronze table, iterator full_refresh + replace writes stay on one WAP branch and publish to main once, after terminal exhaustion. A final merge conflict preserves the branch for inspection. A first sync into a table that does not exist cannot create an Iceberg branch and still uses the direct-to-main bootstrap path, so strict atomic visibility begins with the next replacement. Activity retries may re-fetch the last completed range, but snapshot idempotency prevents duplicate physical rows.

Data envelope

Every extracted record is wrapped with platform metadata before it lands in Iceberg. The envelope provides stable operational fields for dedupe, lineage, sync timing, stream identity, and later transform cursors.

The default transform cursor is _extracted_at; the default tiebreaker is _record_id.

Iceberg naming

All source writes go through Frank's ADL-006 naming helpers. The important rule: do not hand-build Iceberg paths in connector or transform code. Let the naming service combine tenant ID, source name, stream namespace, stream name, and target config.

Typical layers:

• Bronze: raw/source data from source syncs.
• Silver: cleaned and conformed transform outputs.
• Gold: curated business outputs and semantic backing datasets.

Discovery flow

1. Create a source from a pattern.
2. Trigger discovery.
3. Source worker executes discovery through Airbyte or dlt.
4. Frank stores discovered schema and candidate streams.
5. User selects streams and sets sync config.

CLI:

frankctl sources create -f source.yaml
frankctl sources discover <source-id> --timeout 300
frankctl sources streams list <source-id>
frankctl sources streams set <source-id> -f streams.yaml

API:

POST /api/v1/sources
POST /api/v1/sources/{source_id}/discover
GET  /api/v1/sources/{source_id}/streams
POST /api/v1/sources/{source_id}/streams/bulk

Sync flow

1. User triggers a sync manually or through a schedule.
2. Source status moves to syncing.
3. Temporal dispatches work to the source worker.
4. Engine extracts selected streams and writes batches to Iceberg.
5. Sync run records row counts, logs, cursor values, snapshots, and failures.
6. Source status becomes active or error.

CLI:

frankctl sources sync <source-id> --streams customers,orders --timeout 600
frankctl sources history <source-id>
frankctl sources logs <source-id> <run-id>

Schedules

Sources support manual, cron, and interval schedules. These schedules are owned by Temporal and control extraction into Bronze only. They do not run a Pipeline unless that Pipeline separately uses eager or its own clock policy. The CLI exposes Source-scoped schedules:

frankctl schedules set <source-id> -f schedule.yaml
frankctl schedules pause <source-id>
frankctl schedules resume <source-id>
frankctl schedules trigger <source-id>

Example:

schedule_type: cron
schedule_value: "0 */6 * * *"

For the downstream contract and Pipeline schedule commands, see Pipelines: Scheduling and execution ownership.

Failure behavior

If a source fails, it moves to error with last_error_message and last_error_at. Downstream transforms are not deleted or invalidated; they may continue to run against stale tables or block based on their own readiness checks.

That separation is intentional. Fix credentials, rerun discovery if the schema changed, refresh stream schemas, and then sync again.