siphon

Siphon

Siphon is a tool used to replicate data from a PostgreSQL database to other data stores by reading the PostgreSQL logical replication stream. Siphon also takes care of taking the initial data snapshot while the replication stream is being processed.

Siphon has two main components:

• Producer: collecting logical replication events and sending them to a pub-sub system (NATS).
• Consumer: reading data from a pub-sub system, processing logical replication events, and applying them to the target data store.

Currently supported data stores (the consumer type):

Data store	type	Docs
ClickHouse	clickhouse	Example 1, ClickHouse consumer configuration
PostgreSQL	postgresql	PostgreSQL Consumer
Snowflake	snowflake	docs/snowflake_consumer.md, docs/snowflake_operations.md
Parquet on Google Cloud Storage or the local filesystem	gcs_parquet	docs/parquet_consumer.md
Parquet on Amazon S3 or an S3-compatible server (MinIO)	s3_parquet	docs/parquet_consumer.md
Demo (prints data out to STDOUT)	demo	Consumer Development

flowchart LR
    PG[(PostgreSQL)]
    
    subgraph Producer["Siphon Producer"]
        LP[Replication Manager]
        SS[Snapshot Manager]
    end
    
    NATS{{"NATS JetStream"}}
    
    subgraph Consumer["Siphon Consumer"]
        EP[Event Processor]
    end
    
    ClickHouse[(ClickHouse)]
    TargetPG[(PostgreSQL)]
    Snowflake[(Snowflake)]
    Parquet[("Parquet files<br/>GCS / S3 / local")]
    Demo[Demo/STDOUT]
    
    %% Data flow connections
    PG -- "LR Stream" --> LP
    PG -- "Initial Data Snapshot" --> SS
    LP -- "CDC Events" --> NATS
    SS -- "Snapshot Data" --> NATS
    NATS --> EP
    EP --> ClickHouse
    EP --> TargetPG
    EP --> Snowflake
    EP --> Parquet
    EP --> Demo

Siphon is used in production to replicate data from GitLab.com to ClickHouse. It is still under active development.

[[TOC]]

Check out our list of issues to see the development direction.

Further documentation:

• docs/replication_and_snapshot_flow.md: how initial snapshots and continuous CDC hand over through the temporary and snapshot streams.
• docs/consistency_check_and_correction.md: source-to-target consistency checks and corrections.
• docs/default_value_reconciler.md: the ClickHouse default-value reconciler.
• docs/snowflake_consumer.md and docs/snowflake_operations.md: Snowflake consumer design and operations.
• docs/parquet_consumer.md: the gcs_parquet and s3_parquet consumers.
• cmd/schema/README.md: config schema validation and Helm values generation.
• simulator/README.md: the local Docker Compose simulator.
• metrics-catalog/mixins/siphon/README.md: the Grafana dashboards mixin.
• clients/rust/siphon-proto/README.md: the Rust protobuf client for consumers outside Go.

Using GDK?

Before going any further, read the how to guide in the GDK repository to get started locally.

Prerequisites

Siphon requires the following:

• A source PostgreSQL server >=14.0 (with wal_level=logical)
• Go 1.27
• NATS JetStream server
• The target data store of each consumer you run:
  • ClickHouse server (clickhouse)
  • Target PostgreSQL server (postgresql)
  • Snowflake account with key-pair authentication (snowflake)
  • GCS bucket, or a local directory (gcs_parquet)
  • Amazon S3 bucket, or an S3-compatible server such as MinIO (s3_parquet)

Configure wal_level on PostgreSQL

To enable logical replication, go to your postgresql.conf and add:

wal_level=logical

Restart PostgreSQL.

Getting started

• Install the required tooling with mise:

mise install

• Build the Go project:

make

Example 1: data replication into ClickHouse

Configure the producer

Copy the sample application config:

cp sample.config.yml config.yml

Configure the database section to point to your PostgreSQL instance. If needed, create a separate database:

CREATE DATABASE test_siphon;

-- Create a test table
CREATE TABLE users (
  id bigint PRIMARY KEY,
  name text NOT NULL,
  username character varying(50) NOT NULL,
  user_type smallint DEFAULT 3 NOT NULL,
  sign_in_count bigint DEFAULT 0 NOT NULL,
  about text DEFAULT '',
  created_at timestamp with time zone NOT NULL,
  updated_at timestamp with time zone NOT NULL
);

Insert a record:

\c test_siphon;

INSERT INTO "public"."users"
("id", "name", "username", "user_type", "sign_in_count", "about", "created_at", "updated_at")
VALUES (123, 'test', 'test', 2, 2, '', '2024-11-14 13:36:49.714704+00', '2024-11-14 13:36:49.714704+00');

Configure the tables-to-sync section:

table_mapping:
  - table: users # postgres table name
    schema: public # postgres table schema name
    subject: users # NATS subject (queue, topic) name

Configure the consumer

The consumer config lives in the same unified config.yml under the consumers array (the sample.config.yml you copied above already contains a clickhouse consumer). Edit that section rather than copying a second file.

Configure the clickhouse section to point to your ClickHouse instance. If needed, create a separate database:

clickhouse-client -u default --password

CREATE DATABASE replicated_db;

Create a database table for storing data from the PostgreSQL users table:

CREATE TABLE siphon_users (
  id Int64 DEFAULT 0,
  name String DEFAULT '',
  username String DEFAULT '',
  user_type Int8 DEFAULT 3,
  sign_in_count Int64 DEFAULT 0,
  about String DEFAULT '',
  created_at DateTime64(6, 'UTC') DEFAULT now(),
  updated_at DateTime64(6, 'UTC') DEFAULT now(),
  _siphon_replicated_at DateTime64(6, 'UTC') DEFAULT NOW(),
  _siphon_deleted Boolean DEFAULT false
)
ENGINE=ReplacingMergeTree(_siphon_replicated_at, _siphon_deleted)
PRIMARY KEY id;

ClickHouse connection settings

Tune ClickHouse server settings at connection time with an optional connection_settings map under the clickhouse block. Values are strings; numeric strings (e.g. "1", "0") are converted to integers automatically. They are merged on top of Siphon's built-in defaults and take precedence on conflicts.

select_sequential_consistency: 1 is on by default. Everything below is only what you'd add or override per deployment.

clickhouse:
  host: clickhouse
  port: 9000
  database: mydb
  connection_settings:
    insert_quorum: "auto"
    insert_quorum_parallel: "0"
    async_insert: "0"

Pick the settings for your deployment:

Deployment	connection_settings to add
ClickHouse Cloud (SharedMergeTree)	none (default select_sequential_consistency is enough; quorum settings are no-ops on Cloud)
Self-managed single-node	none, or select_sequential_consistency: "0" to drop the unnecessary overhead
Self-managed HA (ReplicatedMergeTree)	insert_quorum: "auto", insert_quorum_parallel: "0", async_insert: "0"

On self-managed HA these are what make Siphon's delete queries safe against lagging replicas.

Note: Any ClickHouse server setting can go here, including max_rows_to_read.

Run NATS

Create a nats-server.conf file to configure the max message size in NATS:

max_payload: 64Mb
max_pending: 64Mb
max_connections: 1000

Run NATS JetStream locally with Docker:

docker run --rm -p 4222:4222  -v ./nats-server.conf:/nats-server.conf  nats -D -c /nats-server.conf -js

Run Siphon

A single config.yml holds the producers, consumers, reconcilers, retention managers and consistency checks, so one process runs them all together:

bin/siphon --config config.yml

All components start concurrently inside the same process, so ordering is handled for you. If you want to run a single component (e.g. on separate hosts) you can still use a subcommand against the same config.yml:

Subcommand	Runs
producer	producers only
consumer	consumers only
reconciler	ClickHouse and Snowflake reconcilers only
retention	retention managers only
consistency	consistency checks only

siphon dashboards export --dir <dir> is not a component: it takes no config and writes the bundled Grafana dashboards into <dir>, then exits.

bin/siphon consumer --config config.yml   # consumers only
bin/siphon producer --config config.yml   # producers only

Flags shared by every subcommand:

Flag	Description
--config	Path to a single config file.
--base-config, --cdc-config	A split config: connection settings in one file, CDC components in the other, merged at runtime. Both are required together; see cmd/schema/README.md.
--log-level	Log level, default info.
--profile	Enable the pprof endpoints under /debug/pprof (see Profiling).

Profiling

Pass --profile to expose the pprof endpoints under /debug/pprof/ on the shared HTTP port (http.port, default 8081). They stay there even when api.port moves the /v1 routes elsewhere. The flag is persistent, so it works with any subcommand:

bin/siphon --config config.yml --profile
bin/siphon consumer --config config.yml --profile

Then point pprof at the running process. The Prometheus metrics tell you that memory or CPU is growing; these tell you which code is responsible:

go tool pprof -http=:8080 http://localhost:8081/debug/pprof/heap
go tool pprof -http=:8080 http://localhost:8081/debug/pprof/profile?seconds=30
curl -s http://localhost:8081/debug/pprof/goroutine?debug=1

For a deployment where the port is not published, forward it first — for example kubectl port-forward pod/<siphon-pod> 8081:8081.

Profiling is off by default and should stay off unless it is needed: /debug/pprof/cmdline exposes the process arguments, heap profiles can contain fragments of replicated row data, and these endpoints carry no authentication — the same caveat as the /v1 endpoints. Unlike /v1, they cannot be moved off the shared HTTP port, which is a reason to leave --profile off in any deployment that publishes that port.

Verification

Query the ClickHouse database using clickhouse-client:

clickhouse-client -d replicated_db -u default --password

SELECT * FROM siphon_users FINAL ORDER BY id

Go to the psql console and update the record:

UPDATE users SET name = 'New Name' WHERE id = 123;

Go back to the ClickHouse database console and verify that the record was updated:

SELECT * FROM siphon_users FINAL ORDER BY id

Insert a new record into PostgreSQL:

INSERT INTO "public"."users"
("id", "name", "username", "user_type", "sign_in_count", "about", "created_at", "updated_at")
VALUES (321, 'NEW USER', 'new_user', 2, 2, '', '2024-11-14 13:36:49.714704+00', '2024-11-14 13:36:49.714704+00');

Verify that the record is present in ClickHouse:

SELECT * FROM siphon_users FINAL ORDER BY id

Example 2: data replication into ClickHouse from GDK

If you've already set up ClickHouse with your GDK you can replicate a few tables to ClickHouse using already created ClickHouse database tables (table schema is already present in Clickhouse).

• users
• projects
• namespaces

Within the GDK/gitlab folder, ensure that you've migrated the ClickHouse database.

bundle exec rake gitlab:clickhouse:migrate

Configure the producer

Configure your producer to use the same database (main DB) and credentials as your GDK/gitlab/config/database.yml in your GDK.

You can reuse the steps from example 1 with a different table_mapping section:

table_mapping:
  - table: users
    schema: public
    subject: users
  - table: projects
    schema: public
    subject: projects
  - table: namespaces
    schema: public
    subject: namespaces

Configure the consumer

Ensure that the clickhouse section of your config.yml points to the ClickHouse database configured in the GDK/gitlab/config/click_house.yml. You can reuse the steps from example 1 with a different streams section:

streams:
  - identifier: "users"
    subject: "users"
    target: "siphon_users"
  - identifier: "projects"
    subject: "projects"
    target: "siphon_projects"
  - identifier: "namespaces"
    subject: "namespaces"
    target: "siphon_namespaces"

Run Siphon

bin/siphon --config config.yml

Verification

After some time you can verify the counts in ClickHouse:

SELECT COUNT(*) from siphon_users FINAL;
SELECT COUNT(*) from siphon_projects FINAL;
SELECT COUNT(*) from siphon_namespaces FINAL;

Example 3: data replication into Parquet files (GCS or S3)

The gcs_parquet and s3_parquet consumers write CDC events as Parquet files with schema-v{version_hash}.json sidecars. See docs/parquet_consumer.md for configuration, AWS authentication and a local MinIO setup (simulator/distributed). Siphon only writes the Parquet files; loading them into Apache Iceberg is done by a separate PySpark job outside this repository.

Cleanup

If you don't intend to use Siphon or just want to restart the applications with a clear state you can follow these steps:

• Stop the producer and consumer applications.
• Delete the PostgreSQL publication (DROP PUBLICATION my_publication, publication_name setting).
• Delete the PostgreSQL replication slot (SELECT pg_drop_replication_slot('my_slot'), slot_name setting).
• Delete the NATS streams (exit docker or use NATS CLI for deleting the streams which contain the configured stream_name substring).

List of configuration options

Producer configuration

path	description
application_identifier	Unique application identifier for the producer. This setting can be used for querying metrics or logs when multiple producers are running. This identifier is used for creating NATS streams, the allowed characters are: letters, numbers, - and _
max_column_size_in_bytes	Maximum column size we ingest from PostgreSQL. This upper limit applies to VARCHAR and TEXT column types. If a column is over the specified size, the producer application replaces it with an empty string. Ideally, this value should be lower than memory_buffer_size_in_bytes.
partitions_monitoring_interval_in_seconds	How frequently should Siphon check for new partitions when a partitioned table is specified in the configuration?
If Siphon detects a partition that is not currently being handled (i.e., not included in the PostgreSQL publication), it will automatically restart. Upon restart, the new partition will be picked up and replicated.
database{}.host	IP or hostname for the PostgreSQL database where the publication and replication slot will be created.
database{}.port	Optional port for the database.
database{}.user	Optional user for the database.
database{}.password	Optional password for the database.
database{}.database	Name of the database.
database{}.advisory_lock_id	The application will take an advisory lock during its runtime to prevent concurrent access in case more than one application instance with the same configuration is started. Each Siphon producer application must have a unique advisory_lock_id value. Default: derived from a hash of application_identifier.
database{}.advisory_lock_timeout_ms	Lock timeout when trying to acquire the advisory lock. The application will automatically retry if the lock couldn't be taken. Default: 100.
database{}.advisory_lock_timeout_fuzziness_ms	Random value in the given range added to the advisory_lock_timeout_ms value to reduce lock contention. Default: 50.
database{}.lock_timeout_ms	Lock timeout value when altering the publication state (ALTER PUBLICATION). The application will automatically retry if the statement fails with a lock timeout. Default: 500.
database{}.lock_timeout_fuzziness_ms	Random value in the given range added to the lock_timeout_ms value to reduce lock contention. Default: 300.
database{}.application_name	Name reported in pg_stat_activity for the producer's connections (suffixed with _replication and _advisory_lock). Default: siphon.
database{}.disallow_hot_standby_feedback_on	When enabled, Siphon will block the snapshotting process for databases that have hot_standby_feedback=on. This safety mechanism is designed to protect the database cluster from potential issues caused by misconfiguration. In production environments, the snapshot process should target a replica where hot_standby_feedback is set to off.
snapshot_database{}	Optional, only the connection settings (host, port, user, password, database, ssl_mode) and disallow_hot_standby_feedback_on are used. This setting can be used to point to a different replica database for doing the initial data snapshot. The main reason for this is avoiding long lock times and affecting the availability of the main DB cluster.
primary_database{}	Optional, only the connection settings (host, port, user, password, database, ssl_mode) are used. This setting can be used when the database option (DB where the replication slot is created) is configured for a replica server. In this case the publication will need to be created on the primary.
replication{}.publication_name	Name of the PostgreSQL publication. This value should be unique for the configured database server.
replication{}.slot_name	Name of the PostgreSQL replication slot. This value should be unique for the configured database server.
replication{}.initial_data_snapshot_threads_per_table	Number of parallel threads to do the initial data snapshot for a table. For large tables, increasing this value is advisable.
replication{}.memory_buffer_size_in_bytes	Size of the memory buffer where the application stores logical replication data before submitting them to the pub-sub system. Default: 3145728 (3 MiB).
replication{}.standby_timeout_ms	How frequently should the application send status updates. The configured value should be lower than the wal_sender_timeout setting in PostgreSQL. Default: 3000ms
replication{}.standby_timeout_emergency_mode_ms	How frequently the application sends status updates while emergency mode is active. Read once when the replication loop starts and used for the duration of that loop; each entry into emergency mode (or back to normal) re-reads the value, so a transition picks up the appropriate setting. Default: 2 × standby_timeout_ms.
replication{}.failover	Boolean flag to use the FAILOVER option when creating the logical replication slot (requires PostgreSQL 17).
replication{}.use_alter_publication_function	Boolean flag to use a function when altering publication, instead of ALTER directly. Default false.
replication{}.snapshot_memory_buffer_size_bytes	Override global replication memory buffer size for snapshots
replication{}.correction_batch_size	Row count a correction range is split into before reading. Packages are cut by the snapshot buffer within each part; a part that hits the statement timeout is read again in halves, and the halved size holds for a minute across the request. Default: 10000.
replication{}.correction_statement_timeout_ms	Statement timeout for the queries a key-range correction runs. A query that hits it is split rather than failing the request; lock_timeout is set to half of it, so a lock wait fails the request for a later retry. 0 disables both. Default: 60000.
replication{}.snapshot_poll_interval_ms	How frequently the snapshot loop polls the snapshot event queue for new snapshot work. Defaults to 3000, minimum 100. Lower values reduce time-to-start for new snapshots; higher values reduce queue load when snapshots are infrequent.
replication{}.snapshot_catch_up_timeout_seconds	How long a snapshot waits for the snapshot database to replay up to the source's current position before reading it. The wait runs once before the table's CDC is moved to the temporary stream and once right after, so a standby snapshot database never serves a state older than CDC already on the main stream; without it a replica lagging behind the producer lets older snapshot rows overwrite newer changes. It compares WAL positions, so it does not cover a snapshot taken on a primary under synchronous replication, which can stream a commit before it is visible there. Hitting the timeout fails the snapshot as retryable, so set it above the snapshot database's worst lag: each wait chases the source's current position, and a replica that is always further behind never passes. Default: 600.
replication{}.snapshot_max_slot_lag_bytes	A new snapshot is not started while the replication slot (on database) lags more than this many bytes behind the database's current WAL position. Every snapshot's merge pauses replication, so starting one on a slot that is already behind adds to the lag. The event stays queued until the lag drops. A snapshot is held for two further reasons, neither of which this setting controls: the slot has not advanced since the last snapshot was allowed, which is merges pausing replication again before it has confirmed anything, and the slot could not be read at all, which does not include a standby that has replayed nothing yet, since nothing it has not replayed can be WAL the slot is holding. The lag is read at most once a minute; the progress check needs no query, since the producer knows what position it last sent. siphon_snapshot_slot_lag_held carries a reason label of lag, no_progress or unreadable, so an expected backlog reads differently from a database that cannot be reached. Default: 5368709120 (5GiB).
replication{}.heartbeat_interval_seconds	How frequently the producer emits a HEARTBEAT event package to each configured stream. Heartbeats flow through the full pipeline (producer → NATS → consumer) and update the siphon_data_lag_ms metric even when there is no real CDC traffic, making it possible to distinguish an idle pipeline from a stuck one. Heartbeats are labelled with the subject's schema/table (the parent table for partitioned tables, since table_mapping isn't partition-expanded), and the metric carries a source label (heartbeat vs data) so heartbeat-driven freshness can be told apart from real end-to-end lag. Default: 60. Set to a negative value (e.g. -1) to disable.
heartbeat_consumer_refresh_seconds	How frequently the producer re-discovers the set of live consumers (via NATS CONSUMER.LIST) to decide which subjects need heartbeats. The consumer set changes rarely, so this is decoupled from heartbeat_interval_seconds and the discovered set is cached and reused between refreshes, keeping the listing load off the NATS cluster. Consumers added after startup begin receiving heartbeats within one refresh window. Default: 300.
replication{}.flush_parallelism	Maximum number of stream keys whose buffered events are submitted to the sink (NATS or emergency object store) concurrently on each flush. Slices sharing the same stream key remain sequential to preserve order across ALTER TABLE version boundaries; only distinct keys parallelize. Default: 2. Increase to overlap publish latency and CPU (proto marshal + zstd) when a flush spans many tables. Tracked by the siphon_flush_buffer_duration_ms metric.
replication{}.preserve_postgresql_timestamp_representation_as_string	When true, date, time, timestamp, and timestamptz column values are stored as raw PostgreSQL string literals (e.g. "4714-11-24 00:00:00+00 BC") instead of being parsed into Go time.Time values. This preserves BC-era dates and original timezone offsets that would otherwise be lost. Applies to both CDC events and initial snapshot data. Default: false. Note: only useful when the target data store accepts string timestamps (e.g. another PostgreSQL instance). Not recommended for ClickHouse consumers.
replication{}.preserve_postgresql_array_representation_as_string	When true, array columns are stored as the raw PostgreSQL literal (e.g. "{a,NULL,b}") instead of the protobuf array types, preserving NULL elements and multi-dimensional arrays. Default: false. Note: for PostgreSQL targets only; not recommended for ClickHouse consumers.
queueing{}.driver	What queueing/pub-sub mechanism to use. Currently nats is supported.
queueing{}.url	URL to the NATS server. Use a tls:// scheme for TLS connections.
queueing{}.tls_config{}.ca_cert_path	Path to the CA certificate file for verifying the NATS server's TLS certificate.
queueing{}.tls_config{}.client_cert_path	Path to the client certificate file for mTLS authentication. Must be provided together with client_key_path.
queueing{}.tls_config{}.client_key_path	Path to the client private key file for mTLS authentication. Must be provided together with client_cert_path.
queueing{}.nats_config{}.replicas	Number of stream replicas in the NATS cluster. Defaults to 1 when unset.
queueing{}.nats_config{}.max_age_seconds	Maximum age of messages in the stream, in seconds. Messages older than this are automatically removed. Defaults to 0 (unlimited / keep forever).
queueing{}.nats_config{}.oversize_bucket_ttl_seconds	Time-to-live for objects in the oversize bucket, in seconds. Objects older than this are automatically removed. Defaults to 0 (unlimited / keep forever).
queueing{}.stream_name	Name of the stream where logical replication events will be submitted. This value should match with the consumer's queueing configuration.
queueing{}.temp_stream_name	Name of the temporary stream for buffering logical replication events for tables which are being snapshotted. Default: temp_<stream_name>
queueing{}.snapshot_stream_name	Name of the snapshot stream for buffering data from the snapshot manager. Default: snapshot_<stream_name>
queueing{}.snapshot_requests_stream_name	Name of the stream snapshot requests are published to and consumed from (<name>.<application_identifier>.snapshot.<snapshot_subject_name>). Created when missing. Setting a dedicated stream is recommended on busy main streams: fetching a rare request from the main stream makes NATS scan every block written since the previous request, which stalls writes to the stream. Must differ from the temp and snapshot streams. Default: <stream_name>
queueing{}.snapshot_status_enabled	When true, Siphon records the lifecycle (in_progress/complete/failed) of each per-table snapshot into a JetStream KV bucket so external consumers (e.g. the org-mover) can poll snapshot progress. Status writes are best-effort and never fail a snapshot. Disabled by default; leave it off unless a consumer reads the bucket.
queueing{}.snapshot_status_bucket_name	Name of the KV bucket holding snapshot status records. Only used when snapshot_status_enabled is true. Default: <stream_name>-snapshot-status
queueing{}.oversize_limit_bytes	Maximum size for events in a message payload; larger events are stored in object storage. (Default: NATS max payload size).
queueing{}.object_storage_config{}.identifier	Unique name for this storage backend. When the entire object_storage_config section is omitted, defaults to "nats-object-storage".
queueing{}.object_storage_config{}.type	Backend type: gcs, s3, file, or nats. Defaults to "nats" when object_storage_config is omitted.
queueing{}.object_storage_config{}.bucket_name	GCS bucket name, local directory (for file), or NATS ObjectStore bucket name (for nats). Defaults to "repl_events_bucket" when object_storage_config is omitted.
queueing{}.failover_storage{}.storage{}.identifier	Identifier stamped on emergency-mode objects. Required when emergency mode is configured. See "Emergency Mode (NATS unavailable)" below.
queueing{}.failover_storage{}.storage{}.type	Backend type: gcs, s3, or file. When unset, emergency mode is disabled.
queueing{}.failover_storage{}.storage{}.bucket_name	GCS or S3 bucket name, or — for the file backend — the root directory under which emergency objects are written.
queueing{}.failover_storage{}.recovery_poll_seconds	How often the producer probes NATS for recovery while in emergency mode. Default: 30.
queueing{}.failover_storage{}.max_concurrent_drain_subjects	Cap on parallel subject-drain goroutines. Default: 4.
table_mapping[].table	Name of the PostgreSQL table.
table_mapping[].schema	Schema name of the PostgreSQL table.
table_mapping[].subject	Subject/queue/topic name where the logical replication events for the given table are going to be sent.
table_mapping[].renamed_table_name	Optional. Name this table is expected to carry after an in-place rename. Lets Siphon follow the same physical table through the rename with no gap and no re-snapshot. See Renamed tables.
table_mapping[].snapshot_memory_buffer_size_bytes	Override snapshot memory buffer per table. Falls back to global config if not defined.
table_mapping[].column_transformations[]	Optional array for ignoring specific columns from the replication
table_mapping[].column_transformations[].column_name	Name of the column we want to ignore
table_mapping[].column_transformations[].transformation_type	ignore is the only supported option.
table_mapping[].row_filters[].column_name	Which column to filter for.
table_mapping[].row_filters[].values_query	Query for calculating the column values.

ClickHouse consumer configuration

path	description
application_identifier	Unique application identifier for the consumer. This setting can be used for querying metrics or logs when multiple producers are running. This identifier is used for creating NATS subscribers, the allowed characters are: letters, numbers, - and _
max_parallel_workers	Caps how many worker goroutines process streams concurrently. Streams are partitioned across the workers, and each worker round-robins over its own streams. Every stream gets at least one fetch turn per cycle, plus up to 3 extra consecutive turns in proportion to its backlog relative to the busiest stream on that worker, so a lagging stream drains faster without starving the quiet ones. Default: one worker per stream.
merge_packages	Merge consecutive CDC packages with the same columns into one handler call when a stream is backlogged. Default: false.
merge_max_messages	Max messages per merged package. Also sets the fetch size and MaxAckPending. Default: 16.
merge_max_bytes	Max decompressed bytes per merged package. Decoded packages take roughly 8x that in heap, per worker. Oversize and snapshot packages are never merged. Default: 16 MiB.
clickhouse{}.host	IP or hostname for the ClickHouse database.
clickhouse{}.port	Port for the ClickHouse database.
clickhouse{}.username	Username for the ClickHouse database.
clickhouse{}.password	Password for the ClickHouse database.
clickhouse{}.database	Name of the ClickHouse database.
clickhouse{}.max_open_conns	Max connections for use at any time. Default: number of configured streams + 1
clickhouse{}.max_idle_conns	Number of connections to maintain in the connection pool. Default: half of max_open_conns
clickhouse{}.ssl	Enables TLS for the ClickHouse connection. Use port 9440 instead of 9000 when enabled. Default: false
clickhouse{}.insecure_skip_verify	Skips TLS certificate verification. For local/dev against self-signed certs only, never production. Overridden by ca_cert_path when that is set. Ignored unless ssl is true. Default: false
clickhouse{}.min_tls_version	Minimum accepted TLS version, "1.2" or "1.3". Ignored unless ssl is true. Default: "1.2"
clickhouse{}.ca_cert_path	Path to a PEM-encoded CA certificate file used as the TLS trust root instead of the system trust pool. A missing or unparsable file fails at connection setup. Ignored unless ssl is true. Default: unset (system trust pool)
clickhouse{}.dedup_config[]	A list of deduplication configurations for specific streams. Each entry defines how deduplication should be handled for a particular stream.
clickhouse{}.dedup_config[].stream_identifier	The identifier of the stream this deduplication configuration applies to. Must match a streams[].identifier.
clickhouse{}.dedup_config[].dedup_by	Replication of DELETE statements requires a 1:1 primary key match between PostgreSQL and ClickHouse. If your keys are mismatched, manually set this configuration by specifying the PostgreSQL primary key columns. Default: uses primary keys specific for the ClickHouse table.
clickhouse{}.dedup_config[].dedup_by_table	Specifies a different table to use for deduplication queries (e.g., when joining with a temporary table). Falls back to the stream's target table if not specified.
internal_events{}.target_table	Enables per-package processing telemetry written back into this ClickHouse table (see Internal processing events). Top-level (sibling of clickhouse); requires clickhouse{} to be configured. Omit the whole internal_events block to disable the feature.
internal_events{}.max_buffer_size	Max number of buffered events held in memory before the oldest are dropped. Default: 100000.
internal_events{}.submit_interval_ms	How often the background flusher writes buffered events to target_table, in milliseconds. Default: 5000.
queueing{}.driver	What queueing/pub-sub mechanism to use. Currently nats is supported.
queueing{}.url	URL to the NATS server. Use a tls:// scheme for TLS connections.
queueing{}.tls_config{}.ca_cert_path	Path to the CA certificate file for verifying the NATS server's TLS certificate.
queueing{}.tls_config{}.client_cert_path	Path to the client certificate file for mTLS authentication. Must be provided together with client_key_path.
queueing{}.tls_config{}.client_key_path	Path to the client private key file for mTLS authentication. Must be provided together with client_cert_path.
queueing{}.nats_config{}.replicas	Number of stream replicas in the NATS cluster. Defaults to 1 when unset.
queueing{}.nats_config{}.max_age_seconds	Maximum age of messages in the stream, in seconds. Messages older than this are automatically removed. Defaults to 0 (unlimited / keep forever).
queueing{}.nats_config{}.oversize_bucket_ttl_seconds	Time-to-live for objects in the oversize bucket, in seconds. Objects older than this are automatically removed. Defaults to 0 (unlimited / keep forever).
queueing{}.object_storage_config{}.identifier	Unique name for this storage backend. When the entire object_storage_config section is omitted, defaults to "nats-object-storage".
queueing{}.object_storage_config{}.type	Backend type: gcs, s3, file, or nats. Defaults to "nats" when object_storage_config is omitted. Must match the producer's configuration so the consumer can read objects the producer wrote.
queueing{}.object_storage_config{}.bucket_name	GCS bucket name, local directory (for file), or NATS ObjectStore bucket name (for nats). Defaults to "repl_events_bucket" when object_storage_config is omitted. Must point to the same storage as the producer.
queueing{}.stream_name	Name of the stream where logical replication events will be submitted. This value should match with the producer's queueing configuration.
streams[].identifier	Identifier for the stream; this setting will be used for creating the NATS consumer.
streams[].subject	Subject/queue/topic name where the logical replication events for the given table are going to be sent.
streams[].target	Target database table where the replication events will be inserted.
streams[].consumer_name	Optional. Overrides the NATS durable consumer name for this stream. When omitted the name is derived as <stream_name>-<application_identifier>-<identifier>. Forbidden characters: ., *, >, whitespace. Must be unique within the consumer config.

Parquet consumer configuration (gcs_parquet, s3_parquet)

Two consumer types write CDC events as Parquet files: gcs_parquet (local filesystem or Google Cloud Storage) and s3_parquet (Amazon S3 or an S3-compatible server). They share one handler, so files, key layout and schema-v{version_hash}.json sidecars are identical; only the storage backend and its credentials differ. The application_identifier, queueing{} and streams[] fields are the same as for the ClickHouse consumer. Full details, AWS authentication options and a minimal IAM policy are in docs/parquet_consumer.md.

Shared fields, under either gcs_parquet{} or s3_parquet{}:

path	description
base_path	First path segment under the bucket, or the root directory when gcs_parquet runs against the local filesystem.
catalog_name	Optional path segment after base_path. Skipped when empty.
warehouse	Optional path segment after catalog_name. Skipped when empty.
schema	Required. Path segment after warehouse; also recorded in file metadata.
include_tenant_in_path	Adds the producer's application_identifier as a path segment between schema and the table name. The _siphon_tenant_id column is written in every row regardless. Default: false
enable_buffering	Buffers rows in memory per (table, version_hash) and flushes on target_rows_per_file or max_buffer_wait_seconds. Buffered rows are lost if the process crashes after the NATS message was acknowledged. Default: false
target_rows_per_file	Row count that triggers a buffer flush. Default: 100000
max_buffer_wait_seconds	Maximum buffer age before a flush regardless of row count. Default: 300
write_batch_size	Events per Parquet file in direct (unbuffered) mode. Default: 10000
tables{}	Per-table overrides keyed by table name, with column_order[] and primary_key.

gcs_parquet{} backend fields:

path	description
gcs_parquet{}.gcs_bucket	GCS bucket name. When empty, files are written to the local filesystem under base_path.
gcs_parquet{}.gcs_project	GCP project. Required when gcs_bucket is set.
gcs_parquet{}.gcs_key_file	Path to a service-account JSON key. Uses Application Default Credentials when empty.

s3_parquet{} backend fields (all optional except s3_bucket):

path	description
s3_parquet{}.s3_bucket	Required. S3 bucket name.
s3_parquet{}.s3_region	Bucket region. When empty the AWS SDK default chain applies (AWS_REGION, shared config, EC2 instance metadata).
s3_parquet{}.s3_endpoint	Endpoint URL for S3-compatible servers such as MinIO or LocalStack. Also enables path-style addressing. Leave empty for AWS.
s3_parquet{}.s3_access_key_id	Static access key. Omit to use the SDK default credential chain (IAM instance role, EKS IRSA, environment, shared config). Credentials are never logged.
s3_parquet{}.s3_secret_access_key	Required when s3_access_key_id is set.
s3_parquet{}.s3_session_token	Session token for temporary static credentials. Only used together with static keys.
s3_parquet{}.s3_role_arn	Role to assume with sts:AssumeRole on top of the base credentials.
s3_parquet{}.s3_max_attempts	Total request attempts for the AWS SDK standard retryer (exponential backoff with jitter). Default: 3

Prometheus metrics use one family per type: siphon_gcs_parquet_consumer_* for gcs_parquet and siphon_s3_parquet_consumer_* for s3_parquet.

Retention manager configuration

The retention manager periodically reclaims space on a NATS stream by setting its MaxAge from the slowest consumer's position, so fully-consumed messages age out without evicting data any consumer still needs. It is configured under the top-level retention_managers[] array and runs as the siphon retention subcommand (or as part of the combined siphon process). Configure one entry per NATS stream — even when multiple producers publish to the same stream, retention is a per-stream concern, not per-producer.

path	description
retention_managers[].application_identifier	Unique identifier for this retention manager, used for log/metric labels. Allowed characters: letters, numbers, - and _.
retention_managers[].schedule	Cron expression controlling how often the manager runs (standard 5-field cron, e.g. 0 3 * * * for daily at 03:00). Required.
retention_managers[].buffer_days	Days added to the slowest-consumer age when computing MaxAge, keeping a margin beyond the last fully-acked message. 0 means no margin. Must be >= 0. Default: 0.
retention_managers[].min_days	Floor that MaxAge can never go below, bounding worst-case data loss if the loop stalls. Must be >= 1. Default: 7 when unset.
retention_managers[].queueing{}	NATS connection and target stream for this manager. Reuses the same queueing schema as producers/consumers; the relevant fields are driver, url, tls_config, and stream_name (the stream whose MaxAge is managed).

Monitoring replication slot lag

siphon_data_lag_ms starts timing when the producer buffers a change it has already decoded, so it cannot see WAL that is committed but not yet read. That is where a producer that cannot replicate shows up: crash-looping, waiting for its advisory lock, or falling behind the write rate. Each producer therefore exports the lag of the slot it replicates from, read every 30 seconds:

Metric	Labels	Meaning
siphon_replication_slot_lag_bytes	app_id, slot_name	WAL between the server's current position and the slot's confirmed position.
siphon_wal_position_bytes	app_id	The server's current WAL position (replay position on a standby), in bytes from 0/0.

The two together give the lag in time, without a database exporter:

siphon_replication_slot_lag_bytes
  / on (app_id) group_left
deriv(siphon_wal_position_bytes[5m])

This assumes WAL is written at a steady rate, so it overstates the lag while writes are quiet and understates it during a burst. The WAL position comes from the producer too, so there is no rate while the producer is down, and just after it returns the rate is averaged across the gap and the estimate reads low. A WAL position from a database exporter, which keeps running through a producer outage, avoids that.

The series are dropped, not frozen, while the slot cannot be read or does not exist and once the producer stops, so a stale value never reads as a healthy slot. An alert or SLO on them has to treat a missing series as bad, e.g. with absent(). Only the slot replication currently reads from is reported, not the standbys in slot_priority_list.

Consistency checks

The consistency checker compares a PostgreSQL table against the table it replicates into (ClickHouse or PostgreSQL) and reports the key ranges where the two disagree. It is configured under the top-level consistency_checks[] array and runs as the siphon consistency subcommand (or as part of the combined siphon process).

Config holds only the connections. The work arrives as a request on a NATS subject naming the source table, the targets to compare it against and the verification mode, so a check is asked for at runtime rather than listed in YAML and redeployed:

nats pub 'siphon_consistency.main_to_clickhouse.request' '{
  "verification_mode": "primary_keys",
  "source": {"table_schema": "public", "table_name": "notes"},
  "targets": [{"identifier": "gitlab_ch", "table_name": "siphon_notes"}]
}'

Divergence can also drive a repair. Give the source a correction block and each mismatch range is submitted, as the checker detected it, to the producer's snapshot subject. The producer splits it into packages that fit, re-reads those rows and clears whatever the target holds that the source does not. Off by default.

verification_mode: custom hashes the columns the request names rather than the primary key, which is how a wrong value gets caught: a badly replicated UPDATE leaves both key sets identical, so primary_keys reports it clean.

nats pub 'siphon_consistency.main_to_clickhouse.request' '{
  "verification_mode": "custom",
  "source": {"table_name": "notes"},
  "columns": [{"name": "id"}, {"name": "note"}, {"name": "updated_at"}],
  "targets": [{"identifier": "gitlab_ch", "table_name": "siphon_notes"}]
}'

The primary key has to be in the list, and both sides read the row rather than an index, so it costs considerably more than primary_keys. Types with no canonical form the two engines agree on (numeric, double precision, bytea, time, arrays) are refused by name rather than silently mismatched. A NULL renders as the type's canonical zero on both sides, which is what the consumer writes into a NOT NULL ClickHouse column, so a nullable source column is comparable without the source knowing the target's schema.

A target whose ClickHouse ORDER BY does not lead with the PostgreSQL key needs an index from somewhere, a projection ordered by that key or a lookup_tables companion, or a pass is thousands of full scans. The check warns and runs either way. The companion does not need the columns being hashed, it is used as an index, turning the key range into the main table's own sorting-key tuples so both reads prune, the same two-stage probe the consumer uses for TOAST and deletes.

Partly implemented: primary_keys and custom modes work end to end against ClickHouse. verification_mode: full is refused, since naming every column in custom mode is the same thing. Partitioned tables are checked per child, and there is no watermark yet, so a table under active write can report divergence that resolves itself and, with corrections on, be snapshotted for it.

The two halves of a check communicate only over NATS, so role: source and role: target can be deployed as separate processes with separate configs and no knowledge of each other, a requirement for the cells topology where one process may not hold connections to both databases. Both sides must agree on application_identifier, which is in every subject.

See docs/consistency_check_and_correction.md for how to configure and run it.

Docker image

You can build the Siphon (producer, consumer) docker image with make:

make docker

docker run -it --rm siphon # args
docker run -it --rm siphon consumer # args

Deployment Setup

Siphon can be configured in different ways to accommodate varying availability and stability requirements.

Single Database Setup

In this configuration, Siphon connects directly to the primary PostgreSQL database for all operations. This setup is ideal for small databases with low traffic. However, the snapshot process may impact database stability as it opens long-running transactions, potentially causing lock contention issues. To minimize disruption, it is recommended to add new tables during off-peak hours.

Replica Database Setup

With PostgreSQL 17, replication slots can be created on replica servers. In this setup, Siphon connects to the primary database only to manage the publication, while both replication and snapshot processes run on the replica. If PostgreSQL 17 is not available, replication must be performed on the primary database instead.

Replica Database Setup with a Dedicated Snapshot Database

In this configuration, Siphon runs the snapshot process from a separate replica node. This node can be configured to not receive user traffic, serving solely for snapshotting data, thereby reducing the impact on production workloads.

Multi-producer setup

When using multiple producers, you may use the same NATS stream for the stream_name setting. However, in that case you must define different temp (temp_stream_name) and snapshot (snapshot_stream_name) stream names for each producer in order prevent collision when these streams are merged to the main (stream_name) after the initial data snapshot for a given table finished.

Following the least privilege model

By default Siphon will attempt to CREATE and later ALTER the publication which requires superuser privileges. On production systems, using Siphon with a user (with superuser privileges) may be a security risk. Following the least privilege model, Siphon can be configured to operate with restricted database users.

The following example will assume a production setup where we have 3 PostgreSQL databases (see Replica Database Setup with a Dedicated Snapshot Database):

• Primary: managing the publication.
• Replica: creating the replication slot and reading the logical replication stream.
• Replica for snapshotting: reading data for the initial data snapshot.

Prerequisites

1. Siphon producer YAML must be configured with the replication.use_alter_publication_function: true setting.
2. A superuser must create the publication manually. The publication name must match the configured replication.publication_name variable.
   • Example: CREATE PUBLICATION my_publication;
3. A special, restricted SQL function must be installed:

CREATE OR REPLACE FUNCTION siphon_alter_publication(pbl TEXT, tbl TEXT, op INTEGER)
RETURNS void AS $$
DECLARE
  operation TEXT;
BEGIN
  IF pbl !~ '^[a-zA-Z_][a-zA-Z0-9_]*$' THEN
    RAISE EXCEPTION 'Invalid publication name';
  END IF;

  IF tbl !~ '^[a-zA-Z_][a-zA-Z0-9_]*\.[a-zA-Z_][a-zA-Z0-9_]*$' THEN
    RAISE EXCEPTION 'Invalid table name format: must be schema-qualified (e.g., public.users)';
  END IF;

  IF op = 0 THEN
    operation := 'ADD';
  ELSIF op = 1 THEN
    operation := 'DROP';
  ELSE
    RAISE EXCEPTION 'Invalid operation parameter: op must be 0 (ADD) or 1 (DROP)';
  END IF;
  
  EXECUTE pg_catalog.format('ALTER PUBLICATION %s %s TABLE %s', pbl, operation, tbl);
END;
$$ LANGUAGE plpgsql SECURITY DEFINER;

Revoke permission to call this function:

REVOKE EXECUTE ON FUNCTION siphon_alter_publication(pbl TEXT, tbl TEXT, operation INTEGER) FROM PUBLIC;

Primary database user configuration

Create a user with read-only privileges:

CREATE USER siphon WITH PASSWORD 'password' NOSUPERUSER NOCREATEDB NOCREATEROLE NOINHERIT;

REVOKE CREATE ON SCHEMA public FROM siphon;

Transfer the ownership of the publication:

ALTER PUBLICATION my_publication OWNER TO siphon;

Allow siphon user to invoke siphon_alter_publication:

GRANT EXECUTE ON FUNCTION siphon_alter_publication(pbl TEXT, tbl TEXT, operation INTEGER) TO siphon;

Configure the primary_database section and after that Siphon will be able to alter the publication.

Replica database user configuration

Create a user with the replica role:

CREATE USER siphon_replicator WITH REPLICATION LOGIN PASSWORD 'password' NOSUPERUSER NOCREATEDB NOCREATEROLE NOINHERIT;

Configure the database section and after that Siphon will be able to create the replication slot and read the logical replication stream.

Snapshot database user configuration

Create a user with read only access:

CREATE USER siphon_snapshot WITH PASSWORD 'password' NOSUPERUSER NOCREATEDB NOCREATEROLE NOINHERIT;

Grant SELECT permission:

GRANT USAGE ON SCHEMA public TO siphon_snapshot;
GRANT SELECT ON ALL TABLES IN SCHEMA public TO siphon_snapshot

Configure the snapshot_database section and after that Siphon will be able to take snapshot from the configured database tables.

ClickHouse Consumer

The ClickHouse consumer inserts logical replication events to the designated database tables. The ClickHouse database tables must follow the following design rules:

• The table engine must be ReplacingMergeTree(_siphon_replicated_at, _siphon_deleted).
• The primary key (or keys) must be the same as the PostgreSQL table.
• The table must have a version column: _siphon_replicated_at DateTime64(6, 'UTC') DEFAULT now().
• The table must have a deleted column: _siphon_deleted Bool DEFAULT FALSE.
• The table can have extra columns or it can have missing columns (not the primary keys).
• The ingestion logic takes the set of available columns in the replication events and the available columns in the ClickHouse table. This logic prevents unknown column errors in ClickHouse.
• The column types should match with the PG types, see the table mapping below.

Data Parsing and Casting

Siphon transforms data through several layers. The following table describes how PostgreSQL types are handled by the producer, represented in Protobuf, and finally mapped to ClickHouse.

PostgreSQL type	Go Parsed Type	In Transit (Protobuf)	Expected ClickHouse Type
bigint	int64	Int64Value	Int64
bigint[]	[]int64	Int64ArrayValue	Array(Int64)
boolean	bool	BooleanValue	Boolean
bytea	[]byte	BytesValue	String
character varying	string	StringValue	String
character varying[]	[]string	TextArrayValue	Array(String)
date	time.Time	TimestampValue	Date32
daterange	string	StringValue	String
double precision	float64	DoubleValue	Float64 or Decimal64
inet	string	StringValue	String
integer	int64	Int64Value	Int64
integer[]	[]int64	Int64ArrayValue	Array(Int64)
int8range	string	StringValue	String
jsonb	string	StringValue	String or JSON
numeric	string	StringValue	Decimal64
smallint	int32	Int16Value	Int16
smallint[]	[]int32	Int16ArrayValue	Array(Int16)
text	string	StringValue	String
text[]	[]string	TextArrayValue	Array(String)
time without time zone	time.Time	TimestampValue	DateTime64
timestamp with time zone	time.Time	TimestampValue	DateTime64
timestamp without time zone	time.Time	TimestampValue	DateTime64
tsvector	string	StringValue	String
uuid	string	StringValue	UUID or String

Note on Numeric Types: PostgreSQL numeric types are sent as strings to preserve full decimal precision during transit. Siphon relies on the target data store to cast these strings to the appropriate numeric or decimal types. double precision types are sent as standard 64-bit floats.

Consistency

The ReplacingMergeTree engine ensures that row changes are going to be eventually consistent; all changes (INSERT/DELETE/UPDATE) will be invoked as INSERT statements using the following rules:

• Depending on the order of events, each row will get its own generated _siphon_replicated_at timestamp value.
• For DELETE operations the _siphon_deleted column will be set to true.

You can see the consistent view of the table using the FINAL keyword (avoid on production-like environments):

SELECT * FROM snippets FINAL

Table Generator

For creating Siphon-ready ClickHouse tables for existing PostgreSQL tables, use the generator in the GitLab Rails project:

bundle exec rails generate gitlab:click_house:siphon snippets

The generator creates a new ClickHouse migration with the CREATE TABLE statement:

class CreateSiphonSnippets < ClickHouse::Migration
  def up
    execute <<-SQL
      CREATE TABLE IF NOT EXISTS siphon_snippets
      (
        id Int64,
        title Nullable(String),
        content Nullable(String),
        author_id Int64,
        project_id Nullable(Int64),
        created_at Nullable(DateTime64(6, 'UTC')),
        updated_at Nullable(DateTime64(6, 'UTC')),
        file_name Nullable(String),
        type Nullable(String),
        visibility_level Int64 DEFAULT 0,
        title_html Nullable(String),
        content_html Nullable(String),
        cached_markdown_version Nullable(Int64),
        description Nullable(String),
        description_html Nullable(String),
        encrypted_secret_token Nullable(String),
        encrypted_secret_token_iv Nullable(String),
        secret Bool DEFAULT false,
        repository_read_only Bool DEFAULT false,
        imported_from Int8 DEFAULT 0,
        organization_id Nullable(Int64),
        _siphon_replicated_at DateTime64(6, 'UTC') DEFAULT now(),
        _siphon_deleted Bool DEFAULT FALSE
      )
      ENGINE = ReplacingMergeTree(_siphon_replicated_at, _siphon_deleted)
      PRIMARY KEY id
    SQL
  end

  def down
    execute <<-SQL
      DROP TABLE IF EXISTS siphon_snippets
    SQL
  end
end

Note: The generator does its best to translate the table schema to ClickHouse; however, a thorough review by a ClickHouse expert is still recommended. If a column type cannot be translated to a ClickHouse type, the generator will place the INSERT_CH_TYPE string in the CREATE TABLE statement.

Internal processing events

The ClickHouse consumer can optionally emit per-package processing telemetry back into a ClickHouse table. After each package is processed, the consumer records a metadata event (producer/consumer identifiers, PostgreSQL schema/table, produced/consumed timestamps, event count, and event type) into a thread-safe in-memory buffer. A single background goroutine periodically flushes the buffer into the target table.

This is useful for querying end-to-end lag, throughput, and event-type breakdowns in SQL rather than only through Prometheus.

Notes:

• The feature is opt-in. Omit the internal_events block to disable it entirely (zero overhead).
• Data loss on crash/shutdown is accepted. On buffer overflow the oldest events are dropped.
• The target table is assumed to already exist; the consumer does not create it. Use the CREATE TABLE statement below.
• The consumer reuses its existing ClickHouse connection and table metadata cache, so the INSERT matches the target table's actual columns.

event_type values:

value	meaning
1	Standard CDC package event
2	Initial snapshot data (first event carries the snapshot operation)
3	Snapshot finished
4	ClickHouse refresh event
5	Consistency correction package

Enable it at the top level of the consumer config (a sibling of clickhouse). The clickhouse{} block must be configured, since the events are written there; startup fails otherwise:

clickhouse:
  host: localhost
  port: 9000
  database: siphon

internal_events:
  target_table: siphon_internal_events
  max_buffer_size: 100000   # optional, default 100000
  submit_interval_ms: 5000  # optional, default 5000

Create the target table before enabling the feature:

CREATE TABLE IF NOT EXISTS siphon_internal_events
(
  uuid UUID,
  stream_identifier LowCardinality(String),
  producer_application_identifier LowCardinality(String),
  consumer_application_identifier LowCardinality(String),
  postgresql_schema LowCardinality(String),
  postgresql_table LowCardinality(String),
  timestamp DateTime64(6, 'UTC') DEFAULT now64(6, 'UTC'),
  produced_at DateTime64(6, 'UTC'),
  consumed_at DateTime64(6, 'UTC'),
  package_events_count UInt64,
  package_size_in_bytes UInt64,
  event_type UInt8,
  version DateTime64(6, 'UTC') DEFAULT now64(6, 'UTC'),
  deleted Bool DEFAULT FALSE
)
ENGINE = ReplacingMergeTree(version, deleted)
PRIMARY KEY (postgresql_schema, postgresql_table, timestamp, uuid)
PARTITION BY toYYYYMM(timestamp)
TTL timestamp + INTERVAL 12 MONTH;

Keeping Related Rows Fresh with refresh_on_change

The Problem

ClickHouse ReplacingMergeTree tables deduplicate by _siphon_replicated_at: the row with the highest timestamp wins. This works well for direct CDC replication, but causes a consistency gap when you use Materialized Views that join multiple tables.

Consider this 1:n example: issues and issue_assignees (one issue has many assignees). A Materialized View denormalises both tables together, for instance, to produce a flat view of issues with their current assignees (as an array column). When an assignee row is added or removed, the parent issues row must be re-processed with a newer timestamp so the Materialized View reflects the change. Without this, the MV continues to show stale data until the issues row is next updated by a direct CDC event.

refresh_on_change solves this by watching a source stream for changes, extracting foreign key values from those events, and then re-inserting the corresponding rows in the related target table with a timestamp that is one microsecond ahead of the current maximum. This guarantees the re-inserted row supersedes whatever version was already in ClickHouse, while remaining behind any CDC events that arrive afterwards.

How it Works

When the consumer processes a batch of events from the source stream:

1. It extracts the foreign key values (configured via source_keys) from each event.
2. It publishes a refresh package to NATS on the subject <stream_name>.<target_stream_identifier>.refresh_packages.
3. A consumer (either the same one, or a dedicated one) receives the refresh package and executes:

INSERT INTO siphon_issues (id, title, ..., _siphon_replicated_at, _siphon_deleted)
SELECT
  id, title, ...,
  addMicroseconds(max_pk_ts, 1) AS _siphon_replicated_at,
  current_deleted AS _siphon_deleted
FROM (
  SELECT
    `siphon_issues`.`id` AS `id`,
    tupleElement(argMax(tuple(`siphon_issues`.`title`, ..., `siphon_issues`.`_siphon_deleted`),
                        `siphon_issues`.`_siphon_replicated_at`) AS _siphon_latest, 1) AS `title`,
    ...,
    _siphon_latest.N AS `current_deleted`,
    max(`siphon_issues`.`_siphon_replicated_at`) AS max_pk_ts
  FROM `siphon_issues`
  WHERE (`siphon_issues`.`id`) IN ((?), (?), (?))
  GROUP BY `siphon_issues`.`id`
)

The latest version of the row is read through one argMax over a tuple rather than one argMax per column. argMax skips NULL values, so a column that was set to NULL would otherwise come back with its previous value, and separate instances can tie-break onto different physical rows.

The reinserted row is versioned at the newest version it observed plus 1µs, not a fresh wall clock: it supersedes what was already in ClickHouse, while a concurrent delete or any CDC event that arrives afterwards still wins.

After executing the reinsert, if the refreshed stream is itself a source_stream_identifier in another refresh_on_change entry, the consumer also emits a further downstream refresh package. This means refresh rules can chain: A → B → C, where a change in A triggers a refresh of B, which in turn triggers a refresh of C.

When target_keys is not the primary key

The reinsert query groups and deduplicates by the target table's primary key, so the refresh package ultimately needs primary key values. When target_keys already are (or include) the primary key, the package is used directly.

When target_keys are a non-primary-key column (but considered somewhat unique on the PostgreSQL side), for example refreshing projects whenever a namespaces row changes:

refresh_on_change:
  - source_stream_identifier: namespaces
    source_keys:
      - id
    target_keys:
      - project_namespace_id   # a foreign key on projects, not its primary key
    target_stream_identifier: projects

the consumer first resolves the primary keys before running the reinsert. It queries the target table for the primary key of every row matching the supplied target_keys values:

SELECT DISTINCT id
FROM siphon_projects
WHERE (project_namespace_id) IN ((42), (43))

and then rebuilds the refresh package keyed by the resolved primary keys (id). A single target_keys value naturally fans out to all matching primary keys (one namespace → many projects). Only integer primary keys are supported for this lookup.

Configuration

Add a refresh_on_change array inside the clickhouse section of the consumer config. Each entry defines a directed refresh rule from a source stream to a target stream:

field	description
source_stream_identifier	The streams[].identifier to watch for changes.
source_keys	Column names in the source stream whose values are the foreign keys to extract.
target_keys	Column names in the target table that the extracted values map to. These are often the target's primary key, but do not have to be (see When target_keys is not the primary key).
target_stream_identifier	The streams[].identifier whose target table should be re-inserted.
filters	Optional list of column / value pairs. Only source events where every column equals its value trigger the refresh. Used to disambiguate polymorphic associations (see Filtering polymorphic sources).

Filtering polymorphic sources

Some source tables are polymorphic: a single foreign key column points at different target tables depending on a discriminator column. For example, label_links joins a label to either an issue or a merge request, with target_id holding the foreign key and target_type saying which table it references ('Issue', 'MergeRequest', ...).

Without filtering, a label_links change would emit a refresh package keyed by target_id against every refresh_on_change rule that watches label_links — refreshing siphon_issues with an ID that's really a merge request ID, and vice versa. filters solves this by restricting each rule to the rows whose discriminator matches:

refresh_on_change:
  - source_stream_identifier: label_links
    source_keys:
      - target_id
    target_keys:
      - id
    target_stream_identifier: issues
    filters:
      - column: target_type
        value: Issue
  - source_stream_identifier: label_links
    source_keys:
      - target_id
    target_keys:
      - id
    target_stream_identifier: merge_requests
    filters:
      - column: target_type
        value: MergeRequest

A few notes on the matching:

• value is a typed YAML scalar, so it can match string discriminators (Issue), integers (5), or booleans (true).
• All filters in a rule must match (AND). An event passes the filter only if every column equals its configured value.
• If the filter column isn't present in the event (not in the package columns, missing, or TOAST-marked), the filter is skipped rather than dropping the event — we'd rather over-refresh than wrongly miss one.

Example Scenario

PostgreSQL schema:

CREATE TABLE issues (
  id        bigint PRIMARY KEY,
  title     text,
  author_id bigint
);

CREATE TABLE issue_assignees (
  issue_id bigint REFERENCES issues(id),
  user_id  bigint,
  PRIMARY KEY (issue_id, user_id)
);

A ClickHouse Materialized View joins siphon_issues and siphon_issue_assignees to build a flat per-issue assignee list. When a row in issue_assignees changes (an assignee is added, removed, or updated), the parent siphon_issues row must be refreshed so the MV re-aggregates.

The refresh direction is: issue_assignees → issues, using issue_assignees.issue_id to look up issues.id.

Producer config — both tables must be in the table mapping:

table_mapping:
  - table: issues
    schema: public
    subject: issues
  - table: issue_assignees
    schema: public
    subject: issue_assignees

Consumer config:

streams:
  - identifier: issues
    subject: issues
    target: siphon_issues
  - identifier: issue_assignees
    subject: issue_assignees
    target: siphon_issue_assignees

clickhouse:
  host: localhost
  port: 9000
  database: replicated_db
  refresh_on_change:
    - source_stream_identifier: issue_assignees
      source_keys:
        - issue_id
      target_keys:
        - id
      target_stream_identifier: issues

When the consumer processes an issue_assignees batch that modifies assignees for issues 10, 11, and 12, it publishes a refresh package that re-inserts siphon_issues rows where id IN (10, 11, 12).

Processing Modes

There are two ways to consume the refresh packages that get published to NATS.

Configure a dedicated stream entry that subscribes to the refresh_packages subject for the target stream. This gives the refresh processing its own NATS consumer position so it cannot stall or be stalled by normal CDC delivery on the issues stream.

streams:
  - identifier: issues
    subject: issues
    target: siphon_issues
  - identifier: issue_assignees
    subject: issue_assignees
    target: siphon_issue_assignees
  # Dedicated entry for refresh packages targeting siphon_issues.
  # The identifier must differ from "issues" to get its own NATS durable consumer.
  - identifier: issues-refresh
    subject: issues.refresh_packages
    target: siphon_issues

The subject follows the pattern <target_stream_identifier>.refresh_packages, which matches the NATS subject the publisher writes to (<stream_name>.<target_stream_identifier>.refresh_packages).

Add enable_refresh_package: true to the target stream. The same consumer that handles issues CDC events will also process the refresh packages for that stream in-line:

streams:
  - identifier: issues
    subject: issues
    target: siphon_issues
    enable_refresh_package: true   # also consume issues.refresh_packages
  - identifier: issue_assignees
    subject: issue_assignees
    target: siphon_issue_assignees

This is simpler to configure but the refresh backlog shares the same NATS consumer position as normal issues CDC events, so a large backlog will delay replication lag for that stream.

Snowflake Consumer

The design and the remaining in-review piece (correction packages) are tracked in issue 304 and written up in docs/snowflake_consumer.md.

The Snowflake consumer (type: snowflake) inserts logical replication events into a pre-existing append-only log table and relies on a Snowflake Dynamic Table (<TABLE>_CURRENT) to compute current state, playing the role ReplacingMergeTree plays for ClickHouse. The consumer only ever runs INSERT; deletes are tombstone rows, a package redelivered by NATS leaves the Dynamic Table's current state unchanged, and types are validated against the packet's column_type_oids and Snowflake INFORMATION_SCHEMA rather than inferred. The Snowflake tables must follow these design rules:

• The log table is a regular table clustered by the primary key: CLUSTER BY (<pk>).
• The primary key (or keys) must be the same as the PostgreSQL table, declared as PRIMARY KEY (...). Snowflake does not enforce it; the consumer reads it via SHOW PRIMARY KEYS. No primary key is a hard error.
• The table must have a version column: _SIPHON_REPLICATED_AT TIMESTAMP_NTZ(6) NOT NULL.
• The table must have a deleted column: _SIPHON_DELETED BOOLEAN NOT NULL DEFAULT FALSE.
• The table can have extra columns or it can have missing columns (not the primary keys). The consumer writes the intersection of the event columns and the table columns.
• The column types should match the PG types; see the mapping table in docs/snowflake_consumer.md.
• Current state is read from <TABLE>_CURRENT. Generate it and the log table with bin/schema generate-snowflake-ddl rather than writing them by hand.

Unchanged TOAST columns in updates are filled from the latest log row, so partial updates never overwrite a value with NULL.

Example for a small snippets table:

CREATE TABLE IF NOT EXISTS "SIPHON"."SNIPPETS" (
    "ID"                    NUMBER(38,0)     NOT NULL,
    "TITLE"                 VARCHAR,
    "CONTENT"               VARCHAR,
    "AUTHOR_ID"             NUMBER(38,0),
    "PROJECT_ID"            NUMBER(38,0),
    "CREATED_AT"            TIMESTAMP_NTZ(6),
    "UPDATED_AT"            TIMESTAMP_NTZ(6),
    "VISIBILITY_LEVEL"      NUMBER(38,0),
    "_SIPHON_REPLICATED_AT" TIMESTAMP_NTZ(6) NOT NULL,
    "_SIPHON_DELETED"       BOOLEAN          NOT NULL DEFAULT FALSE,
    PRIMARY KEY ("ID")
)
CLUSTER BY ("ID");

CREATE OR ALTER DYNAMIC TABLE "SIPHON"."SNIPPETS_CURRENT"
    TARGET_LAG = '5 minutes'
    WAREHOUSE = SIPHON_WH
AS
SELECT *
FROM (
    SELECT *
    FROM "SIPHON"."SNIPPETS"
    QUALIFY ROW_NUMBER() OVER (
        PARTITION BY "ID"
        ORDER BY "_SIPHON_REPLICATED_AT" DESC
    ) = 1
)
WHERE "_SIPHON_DELETED" = FALSE;

Schema changes follow the same runbook as ClickHouse: unless on_schema_drift: evolve is set, apply the target DDL (ALTER TABLE ... ADD COLUMN) before running the PostgreSQL migration. Configuration keys and defaults, the drift policy, metrics, the write path, testing, and open decisions are all in docs/snowflake_consumer.md.

Database Schema Management

Siphon does not enforce a strict mapping between the schemas of the data stores (PostgreSQL, ClickHouse); this must be ensured by the user. By default, Siphon uses best-effort mechanisms to ensure that data is correctly inserted into ClickHouse.

For example, the ClickHouse consumer only inserts columns that are available in ClickHouse, the rest will be ignored.

Handling Database Schema Changes

This process must be managed externally, outside of Siphon's context. Siphon relies on logical replication, which does not replicate database schema changes. The logical replication protocol sends up-to-date table information when a table changes, but there is no way to retrieve the underlying SQL statements that caused the change.

Example:

ALTER TABLE issues RENAME COLUMN iid TO internal_id;

In the replication stream, we wouldn't see the RENAME operation. Instead, we would see that new replication events no longer contain the iid column, and a new column, internal_id, is present.

Static Database Schema

When the database schema does not change after the initial ClickHouse (or other data store) schema is created, there is nothing else to do.

Dynamic Database Schema

Before updating the PostgreSQL database schema, the downstream database schema must be updated to reflect the changes.

Example: adding a new column to the issues table.

ALTER TABLE issues ADD COLUMN author_id bigint;

Before executing this SQL statement in PostgreSQL, the downstream database schema must be updated. In the case of ClickHouse, you would run:

ALTER TABLE siphon_issues ADD COLUMN author_id Int64;

Replicating Partitioned Database Tables

In the producer configuration, you can configure individual table partitions in the table_mapping section.

table_mapping:
  - table: audit_events_2023
    schema: public
    subject: audit_events
  - table: audit_events_2024
    schema: public
    subject: audit_events

As you can see, both partitions send their events to the audit_events subject. Since both tables share the same schema, the downstream database may choose to store all records in a single table. This is because a partitioning scheme that makes sense in PostgreSQL might not be possible or useful in other data stores. Depending on your use case, you may keep the subject values separate.

Dynamic Partitions

The Siphon producer accepts partitioned tables in the table_mapping section. When a partitioned table is specified, Siphon looks up all partitions for the given table and adds them one by one to the publication.

Some partitioned tables change over time as older partitions are dropped and new ones are added. Siphon performs a periodic check that detects partition changes for the configured partitioned tables. This happens automatically every 5 minutes and can be overridden using the partitions_monitoring_interval_in_seconds setting. When a partition change is detected, Siphon reloads itself, adding the new tables to the PostgreSQL publication. The downtime where replication messages are not processed is minimal, just a few seconds at most.

For new tables, Siphon will also take an initial data snapshot to ensure no data is missed.

Information About the VersionHash

The VersionHash is a Protobuf field that acts as a fingerprint. It represents the specific state of a database table (its schema) at the time an event package was generated.

You may use the VersionHash to detect database schema changes ("schema drift") between the PostgreSQL source and your downstream data store.

Ensuring Schema Consistency

To ensure consistency, implement the following logic in your consumer:

1. Baseline Compatibility: Ensure your downstream data store schema is currently compatible with the PostgreSQL source.
2. Extract: When your consumer receives an events package, extract the VersionHash value.
3. Compare: Compare this value against the last known safe VersionHash (stored in a database or state store).

Scenario A: The Hashes Match

• Status: The schema has not changed.
• Action: Safely process the events.

Scenario B: The Hashes Differ

• Status: The PostgreSQL table schema has changed.
• Action:
  1. Halt: Stop processing immediately by returning an error. The consumer goroutine for this table enters into a retry loop.
  2. Inspect: Compare the current PostgreSQL schema against your downstream schema to identify the changes.
  3. Migrate: Apply the necessary changes to the downstream database.
  4. Resume: Once confident in the changes, clear or update the stored VersionHash. The retry loop will succeed on the next attempt, and processing will continue.

How is VersionHash Calculated?

The hash is an integer derived using the XXH3 algorithm. It is calculated based on:

• The Table Name
• The Table Schema
• The list of Columns (sorted alphabetically)

Example:

schema.table_name,column1::type_id,column2::type_id

XXH3 uint64:

337559362357771368

You can get the metadata in the correct format for a given table using the following PostgreSQL query:

-- Example: public.users
SELECT 'YOUR_SCHEMA.YOUR_TABLE,' || string_agg(columns.column, ',')
FROM (
  SELECT
    (a.attname || '::' ||  a.atttypid) AS column
  FROM
    pg_catalog.pg_attribute a
  JOIN
    pg_catalog.pg_class c ON a.attrelid = c.oid
  JOIN
    pg_catalog.pg_namespace n ON c.relnamespace = n.oid
  WHERE
    n.nspname = 'YOUR_SCHEMA'
    AND c.relname = 'YOUR_TABLE'
    AND a.attnum > 0
    AND NOT a.attisdropped
  ORDER BY
    1
) AS columns

Column Type OIDs

CDC and snapshot packages also carry column_type_oids, a list of PostgreSQL type OIDs index-aligned with the columns list (column_type_oids[i] is the OID of columns[i]). Consumers may optionally use these to inspect the source column types (for example, to detect exactly which type changed when the VersionHash differs). The OIDs are the values from pg_type.oid.

Ignoring Sensitive Columns

The Siphon producer can skip specified columns from the replication thus sensitive or not needed data will not going to be sent to the consumers (nor to NATS). On the consumer side, the ignored columns will not going to be part of the INSERT statement (ClickHouse consumer).

Example table mapping configuration:

table_mapping:
  - table: users
    subject: users
    schema: public
    column_transformations:
      - column_name: reset_password_token # sensitive data
        transformation_type: ignore
  - table: issues
    subject: issues
    schema: public
    column_transformations:
      - column_name: description_html # HTML representation is not needed
        transformation_type: ignore

Row Filtering

Instead of taking a snapshot of the whole table, you can select certain rows using the row_filters configuration.

Example: select issue records from a given organization

  - table: issues
    subject: issues
    schema: public
    row_filters:
      - column_name: namespace_id
        values_query: "SELECT id FROM namespaces WHERE organization_id = 1"

This configuration would produce the following snapshot query:

SELECT * FROM issues
WHERE namespace_id IN (SELECT id FROM namespaces WHERE organization_id = 1)

Example: use multiple filters for parallel snapshotting (the filters should not overlap)

  - table: issues
    subject: issues
    schema: public
    row_filters:
      - column_name: namespace_id
        values_query: "SELECT 9970"
      - column_name: project_id
        values_query: "SELECT 1234"

Example: snapshot nothing

  - table: issues
    subject: issues
    schema: public
    row_filters:
      - column_name: id
        values_query: "SELECT NULL::bigint"

Row filters apply to both phases: the initial data snapshot filters at SQL level (WHERE col IN (subquery)), and the logical replication stream checks each INSERT/UPDATE against an in-memory bitmap built at producer boot. Filters on a partitioned table are inherited by its partitions in both phases.

Note: the allow-list is loaded once at boot, so changing what a values_query returns requires restarting the producer. DELETEs are never filtered, see Limitations.

Renamed tables

PostgreSQL tracks publication membership by OID, but Siphon routes events by name. A renamed table therefore keeps producing WAL under a name Siphon no longer recognises, which means dropped events and a forced re-snapshot.

renamed_table_name declares the rename up front so Siphon follows the same physical table through it. Same subject, same schema version hash, no gap, and no re-snapshot:

  - table: issues
    subject: issues
    schema: public
    renamed_table_name: issues_old

Given an in-place swap:

BEGIN;
ALTER TABLE issues_new RENAME TO issues_tmp;
ALTER TABLE issues     RENAME TO issues_old;   -- the table Siphon replicates
ALTER TABLE issues_tmp RENAME TO issues;
COMMIT;

Siphon keeps replicating the table now named issues_old and still publishes it as public.issues, so consumers and ClickHouse see no change at all.

Rules worth knowing:

• The alias is ignored while the configured name is still live in the publication, so it's safe to configure ahead of the rename even when a different table already bears the alias name.
• renamed_table_name must not collide with another mapping's table, or with its own. Config validation rejects that, since events for the two would be indistinguishable by name.
• Partitioned tables need only the parent entry, even when the partitions are renamed too. Siphon publishes the individual partitions, so it derives each one's expected renamed name from the parent's: PostgreSQL names a partition <parent>_<suffix> and a rename carries the suffix across, so p_old_1 and p_new_1 resolve to the same replicated table. New partitions are still discovered automatically. A partition whose name doesn't start with the parent's isn't covered and needs its own entry.
• Siphon follows the OID, so it keeps replicating the table it was already replicating. If a swap's intent is the opposite (retire that table in favour of a rebuilt twin taking over its name), don't use this field: Siphon would follow a table that no longer receives writes. Siphon logs an error when it sees two relation OIDs resolving to one configured table, which is the symptom.
• Don't trigger a manual snapshot for a mapping whose rename is in effect: the snapshot resolves by the configured name, which now belongs to the other table.
• If the alias entry is removed from the publication by something other than Siphon while a decoy holds the configured name, the next reconcile adopts and snapshots the decoy. Siphon's own paths never drop it, so this needs an outside actor (a cleanup script, or another producer sharing the publication whose config lacks the field).
• Which name is live is decided from publication membership, so the field only works for a rename Siphon was already replicating across. Configuring it for a rename that happened while the table was never published (fresh producer, recreated publication) does nothing: Siphon looks for the configured name, warns that it doesn't exist, and skips the mapping. Point table at whichever name actually exists instead.

Snapshot Status API

When queueing.snapshot_status_enabled is true (see producer configuration), the producer records the lifecycle of every per-table snapshot into a JetStream KV bucket and exposes it over HTTP. External systems (for example, the org-mover) can poll this endpoint to learn whether a snapshot has completed without scraping logs.

The endpoint is served by the producer's HTTP server — the same one that serves /metrics and /health, listening on http.port (default 8081), unless api.port moves the /v1 routes to their own listener.

Endpoint

GET /v1/snapshot-status/{schema}/{table}

Parameter	In	Required	Description
schema	path	yes	PostgreSQL schema name (e.g. public).
table	path	yes	PostgreSQL table name (e.g. users).
filters_hash	query	no	Selects a filtered snapshot. Omit it to look up the unfiltered snapshot. See Computing filters_hash.

The response is a JSON array. A single producer returns at most one element; when multiple producers write to the same bucket and each has a record for the key, every match is returned, sorted by app_id.

Field	Description
app_id	application_identifier of the producer that owns the record.
schema / table	The snapshotted table.
filters_hash	Present only for filtered snapshots.
row_filters	The column_name / values_query pairs the snapshot used (filtered snapshots only).
publication_lsn	The publication LSN captured when the snapshot ran.
status	One of in_progress, complete, or failed.
started_at	When the snapshot started (UTC).
completed_at	When it reached a terminal state (complete/failed); absent while in_progress.
error	Failure cause; present only when status is failed.

Status codes:

• 200 OK — one or more records found.
• 400 Bad Request — schema or table is missing from the path.
• 404 Not Found — no producer has a record for the requested key.

The status records are durable (they live in the KV bucket, not in memory), so they survive producer restarts and remain queryable even before any new snapshot runs.

Examples

Unfiltered snapshot:

curl -s http://localhost:8081/v1/snapshot-status/public/users | jq

[
  {
    "app_id": "my_producer",
    "schema": "public",
    "table": "users",
    "publication_lsn": "0/19F16B8",
    "status": "complete",
    "started_at": "2026-05-29T06:23:23.507709Z",
    "completed_at": "2026-05-29T06:23:26.569597Z"
  }
]

Filtered snapshot — pass the hash of the table's row_filters (see below):

curl -s "http://localhost:8081/v1/snapshot-status/public/issues?filters_hash=04322d4838c11259" | jq

[
  {
    "app_id": "my_producer",
    "schema": "public",
    "table": "issues",
    "filters_hash": "04322d4838c11259",
    "row_filters": [
      { "column_name": "namespace_id", "values_query": "SELECT id FROM namespaces WHERE organization_id = 1" }
    ],
    "publication_lsn": "0/19F2A40",
    "status": "complete",
    "started_at": "2026-05-29T06:23:29.723985Z",
    "completed_at": "2026-05-29T06:23:32.902956Z"
  }
]

Computing filters_hash

A snapshot is identified by (schema, table, row_filters). To keep the KV key fixed-length and independent of how the filters are ordered in the config, the configured row_filters are reduced to a 16-character hex digest. A caller that wants to look up a filtered snapshot must compute the same digest from the table's row_filters; an unfiltered snapshot uses the empty hash and the shorter key status.<schema>.<table>.

The digest is computed as follows:

1. Take the table's row_filters — each is a column_name + values_query pair.
2. Sort them by column_name, then by values_query. Filter ordering in the config therefore does not change the hash.
3. For each filter, append column_name, a 0x00 byte, values_query, then a 0x01 byte to a buffer. The separators prevent two different filter sets from colliding into the same byte string.
4. Hash the buffer with XXH3 (64-bit) and format it as a zero-padded, lowercase 16-char hex string (%016x).
5. An empty or absent row_filters list hashes to "" (the unfiltered case).

For example, the single filter from the Row Filtering example —

row_filters:
  - column_name: namespace_id
    values_query: "SELECT id FROM namespaces WHERE organization_id = 1"

— produces the buffer namespace_id\x00SELECT id FROM namespaces WHERE organization_id = 1\x01 and hashes to 04322d4838c11259.

For a single filter you can reproduce the hash on the command line with xxhsum (≥ 0.8.0, where -H3 selects XXH3-64). The \x00 / \x01 bytes are the field/record separators described above:

printf 'namespace_id\x00SELECT id FROM namespaces WHERE organization_id = 1\x01' | xxhsum -H3 | cut -d_ -f2 | cut -d' ' -f1
# => 04322d4838c11259

For multiple filters, sort them by (column_name, values_query) and concatenate each column_name\x00values_query\x01 segment into the same buffer before hashing.

A request using the structured filters instead appends a second section to the same buffer, after the row-filter one, so a request with only row_filters hashes exactly as it did before this existed. Each filter contributes type\x00, then column\x00 per column in order, then one marker byte per bound (\x02 for an omitted or empty bound, \x03 followed by value\x00 per value), then \x01. The marker is what keeps {start: ["1"]} and {end: ["1"]} on different status keys. Filters are sorted by (type, columns, start, end); the columns within a filter are never reordered, since their order is the range's meaning.

The reference implementation is HashSnapshotFilters in pkg/siphon/snapshot_status.go.

Snapshot Trigger API

Starts a snapshot of one or more tables over HTTP, either partial (each table with its own row filters) or full-table — the programmatic replacement for publishing SnapshotEvents to NATS by hand. This is the operational front door for partial snapshots: a caller that has just widened a table's row filter (for example, added namespace IDs 4,5 to an existing 1,2,3), updated the producer config, and restarted the producer uses it to backfill the rows that existed before the change. CDC already streams the new IDs from config.

The snapshot's scope is exactly what the request's row_filters match, and they are independent of config — they fully replace the config filters for this run. CDC, however, keeps replicating by the config filters, so keep the request filters within what config covers: any rows you snapshot beyond it are backfilled once and then drift, because CDC never streams their changes. The widen-and-backfill flow above is the safe case (the request filters are a subset of the just-widened config filters); a full-table snapshot is consistent only when the table has no config filter at all.

Leaving row_filters out (or passing []) triggers a full-table snapshot instead: the table is sliced by page ranges and snapshotted in parallel, exactly like the initial snapshot at boot. Its status record lands under the unfiltered key, so filters_hash comes back empty and the returned status_url polls that key. Same consistency caveat as above: it's only safe when the table has no config row filter.

The endpoint is served by the same HTTP server as /metrics, /health, and the Snapshot Status API, listening on http.port (default 8081), unless api.port moves the /v1 routes to their own listener.

Endpoint

POST /v1/snapshot

The schema, table, and filters all travel in the body, so one request can trigger several tables at once. All tables in a request share a single boundary LSN.

Field	In	Required	Description
tables	body	yes	Non-empty array of tables to snapshot. Each entry: schema, table, and optionally either row_filters or filters (see Structured filters). Omit both for a full-table snapshot. Duplicate schema.table entries are rejected.
app_id	body	no	Targets a specific producer by application_identifier. Auto-selected when exactly one producer is registered; required when more than one is.

Structured filters

row_filters is the original form: column_name / values_query pairs, where values_query is raw SQL interpolated into WHERE column IN (<subquery>). It cannot describe an ordered tuple range and it cannot travel in a package, so it only ever scopes the snapshot query.

filters is the structured form, and a request carrying it produces a correction package instead of a plain snapshot package: the filters are encoded into the package so a consumer can clear exactly the rows the producer snapshotted before inserting them. Use it to repair a diverged key range.

The two are mutually exclusive on one table entry.

Filter field	Required	Description
type	yes	key_range is the only supported kind today.
columns	yes	The key columns, in the order they should be compared. Normally the table's primary key in index order, which is what keeps the scan on the index and what a consumer can scope a delete by. The range is compared lexicographically, so the order is part of its meaning. They have to identify a row: a unique index on NOT NULL columns among them, or the snapshot fails without a retry.
start	no	Inclusive lower bound, one value per column. Omit for a range open at the low end.
end	no	Exclusive upper bound, one value per column. Omit for a range open at the high end.

Exactly one key_range filter is required when filters is present.

Bound values are always strings, in the request and in the package alike, and their types come from the source catalog rather than from the request. That is what lets a bigint primary key above 2^53 survive the trip: as a JSON number it would round silently. The consumer resolves each bound's type from the package's column_type_oids and binds the text form itself.

Every type that appears in a GitLab primary key is supported: smallint, integer, bigint, text, character varying, uuid, boolean, date, timestamp and timestamptz. Timestamps are RFC 3339, and a date also accepts YYYY-MM-DD. Types that never appear in a primary key (numeric, bytea, the array types) are rejected.

Column names and bound values are checked against the table when the snapshot starts, not when the request is accepted: the producer reads the table's shape there anyway, so a second catalog read at request time would buy nothing. A column that is not on the table, or a value that is not representable as its column's type, therefore comes back as a failed snapshot status with the reason rather than as a 400. It is not retried, since no retry could fix it.

A text or character varying key is compared under the source database's default collation, uncollated. Anything generating these ranges has to enumerate the key space the same way, or the correction covers a different row set than the one that was detected and part of the divergence survives while the correction reports success.

Accepted risk: ClickHouse compares String bytewise while PostgreSQL's default collation is locale-dependent. The two agree for ASCII text, which is what primary keys hold in practice, and can disagree beyond it. A text-keyed table reporting divergence across the board is the symptom.

Bounds are passed as query placeholders, never interpolated.

A correction range is published as one or more packages, each covering its own narrower key range, and a range is published even when it matched no rows. That is the common case when the target holds rows the source does not, where clearing the range is the whole fix. The producer splits a range into parts of replication.correction_batch_size rows, reads each in key order, and cuts a package at the row that would overflow the snapshot buffer, so the packages are disjoint and no package's delete reaches another's rows. A part that hits replication.correction_statement_timeout_ms is read again in halves; a single row that still times out is skipped and logged with its bounds, and the snapshot fails only if nothing of the range was published. A lock wait fails the snapshot so it is retried later. When the snapshot DB sits behind a load balancer, a slice connection that lands on a replica without the exported snapshot reconnects, up to 20 times. A correction scoped by a partition filter alone has no key to split on, and fails if it does not fit one package.

Only the ClickHouse consumer applies corrections. It inserts the package's rows first, then tombstones the keys in the range whose newest version predates that insert, so a key the correction carried excludes itself and the range is not rewritten wholesale. A failure between the two leaves stale rows rather than a hole, which the next check pass re-detects. The PostgreSQL, Snowflake and Parquet (gcs_parquet, s3_parquet) consumers log and drop them, since inserting the rows without the clear would leave the target holding exactly the rows the correction was meant to remove.

A complete snapshot status still tells you the producer published the packages, not that a target applied them. siphon_clickhouse_consumer_corrections_total reports that, by outcome.

A partitioned source needs a partition filter beside the key range, and the producer refuses a correction for a partition child without one. A tuple range does not constrain its trailing columns at intermediate values, so a range over (id, partition_id) also matches sibling partitions, which ClickHouse holds in the same table and the correction never snapshotted. partition_in and partition_range scope those out, and the refusal names the partition column to use. The consistency checker emits one automatically.

A key range open at both ends is rejected at request time with a 400: it is the whole table, which an unfiltered request already snapshots, and as a correction it adds only a range delete. The consumer refuses, rather than approximating, anything else it cannot apply exactly: a filter column the target lacks, a filter kind it predates, a uuid bound, or a version column coarser than microseconds. A row versioned at or past the correction's own timestamp is skipped rather than refusing the whole range. It logs and acks in every case, so a correction never blocks the stream.

Bounds are read using the column's PostgreSQL type, taken from the package's column_type_oids, not the ClickHouse column's. The two are not interchangeable: timestamp and timestamp with time zone both map to DateTime64, but PostgreSQL discards an offset-bearing bound's offset for the first and honors it for the second, so reading the ClickHouse type shifts the range and the consumer clears rows the producer never snapshotted. uuid is refused for the same reason in a form that cannot be fixed by reading the bound differently: PostgreSQL orders a uuid by memcmp over its 16 bytes, while ClickHouse holds it as two UInt64s and compares the halves in the opposite order, so the same [start, end) is a different set of rows on each side.

curl -s -XPOST http://localhost:8081/v1/snapshot \
  -H 'Content-Type: application/json' \
  -d '{
        "tables": [
          {"schema":"public","table":"ci_builds_101",
           "filters":[{"type":"key_range",
                       "columns":["id","partition_id"],
                       "start":["1000","101"],
                       "end":["2000","101"]},
                      {"type":"partition_in",
                       "columns":["partition_id"],
                       "values":["101"]}]}
        ]
      }'

Which snapshots exactly:

SELECT ... FROM "public"."ci_builds_101"
WHERE ("id", "partition_id") >= ($1, $2)
  AND ("id", "partition_id") <  ($3, $4)
  AND "partition_id" IN ($5)

The partition filter changes nothing on the source, which reads the child directly; it is there for the consumer, and the producer refuses a correction for a child without it. A request carries at most one key_range and at most one partition filter, AND-ed. partition_range is the RANGE equivalent, taking start/end over the one column, where an absent bound means MINVALUE/MAXVALUE.

The response is 202 Accepted — the events are enqueued, not finished. The snapshots run asynchronously on the producer's snapshot loop; poll each table's returned status_url (the Snapshot Status API) to observe in_progress → complete/failed.

Response field	Description
app_id	The producer the snapshots were routed to.
lsn	The boundary LSN, read server-side at request time and shared by every table in the batch. CDC covers changes after it; each snapshot backfills rows committed as of it; the merge dedupes the overlap.
tables[]	One entry per snapshotted table: schema, table, filters_hash (the filters_hash of that table's filters — matches its status record key), and a ready-to-poll status_url.

Validation is all-or-nothing: if any table is unconfigured or malformed, the whole request is rejected and nothing is enqueued. Re-submitting is safe — an already-snapshotted table is deduped by the merge.

Validation happens at enqueue time, so an event can still outlive its config: if a table is removed from table_mapping (or the producer is restarted with a narrower config) while its snapshot event is still queued, the snapshot loop drops that event and logs a warning. Nothing retries it — table_mapping is only read at startup, so the event could never succeed. Re-add the table and re-trigger the snapshot if you still want it.

Status codes:

• 202 Accepted — events enqueued.
• 400 Bad Request — malformed JSON; empty tables; a table missing schema/table; a row_filters entry missing column_name/values_query; row_filters and filters on one entry; more than one filter; a filters entry with an unknown type, no columns, a duplicate or empty column name, or a bound whose length does not match columns; a duplicate schema.table. An unknown column or an unrepresentable bound value is not caught here, see Structured filters.
• 404 Not Found — a table is not configured on the target producer, a configured table is not part of its publication, or app_id is unknown.
• 409 Conflict — multiple producers are registered and no app_id was given (ambiguous).
• 503 Service Unavailable — no producer is registered (startup or failover mode — snapshotting is suspended in failover).
• 500 Internal Server Error — reading the LSN or enqueueing an event failed.

Example

curl -s -XPOST http://localhost:8081/v1/snapshot \
  -H 'Content-Type: application/json' \
  -d '{
        "tables": [
          {"schema":"public","table":"issues",
           "row_filters":[{"column_name":"namespace_id","values_query":"SELECT id FROM namespaces WHERE id IN (4,5)"}]},
          {"schema":"public","table":"merge_requests",
           "row_filters":[{"column_name":"namespace_id","values_query":"SELECT id FROM namespaces WHERE id IN (4,5)"}]}
        ]
      }' | jq

{
  "app_id": "my_producer",
  "lsn": "1E/D0CA9818",
  "tables": [
    {
      "schema": "public",
      "table": "issues",
      "filters_hash": "04322d4838c11259",
      "status_url": "/v1/snapshot-status/public/issues?filters_hash=04322d4838c11259"
    },
    {
      "schema": "public",
      "table": "merge_requests",
      "filters_hash": "9f1c0b7a2e6d4a83",
      "status_url": "/v1/snapshot-status/public/merge_requests?filters_hash=9f1c0b7a2e6d4a83"
    }
  ]
}

Then poll each table for completion:

curl -s "http://localhost:8081/v1/snapshot-status/public/issues?filters_hash=04322d4838c11259" | jq '.[0].status'
# "in_progress" -> "complete"

Replication Control API

Pauses and resumes snapshotting over HTTP, and reports the producer's state. This is the first half of driving database maintenance without the Siphon team; pausing logical replication itself comes next.

Served by the same HTTP server as /metrics, /health, the Snapshot Status API, and the Snapshot Trigger API, listening on http.port (default 8081). There is no authentication on any /v1 route: keep whichever port serves them network-internal, and see Serving the /v1 API on its own port.

Endpoints

Endpoint	Description
GET /v1/status	Reports the current state, changes nothing.
POST /v1/snapshot/pause	Stops the snapshot loop from picking up new snapshot events, and suspends partition monitoring. An in-flight snapshot runs to completion.
POST /v1/snapshot/resume	Re-enables snapshot processing and partition monitoring.

app_id is an optional query parameter on every endpoint (note: the Snapshot Trigger API takes it in the body instead). It targets a specific producer by application_identifier, and is auto-selected when exactly one producer is registered.

Every endpoint answers with the same body, so a mutating call echoes the resulting state:

{
  "app_id": "my_producer",
  "snapshot_paused": true,
  "snapshot_running": false
}

Status codes:

• 200 OK — current state returned; for mutating calls the change has taken effect.
• 404 Not Found — app_id is unknown.
• 409 Conflict — multiple producers are registered and no app_id was given.
• 503 Service Unavailable — no producer is registered.

Behaviour while snapshotting is paused

• Snapshot triggers still enqueue. POST /v1/snapshot keeps accepting requests; the events wait on the main stream and are picked up on resume.
• Partition monitoring is suspended, because acting on a new partition restarts the producer, which would end the maintenance window. New partitions are picked up on the first check after resume.
• The state is in memory. A restart resumes snapshotting; persistence comes later in this series.
• siphon_snapshot_paused{app_id} is 1 while paused. Nothing auto-resumes, so alert on it.

Serving the /v1 API on its own port

As /v1 is mutating and unauthenticated API, it is better to run it on its own port for separation of concern from the o11y port.

Set api.port to move the /v1 routes to a second listener:

http:
  port: 8081   # /metrics, /health*, /debug/pprof
api:
  port: 8082   # /v1/* only

	api.port unset	api.port set
http.port	/metrics, /health*, /v1/*, pprof	/metrics, /health*, pprof
api.port	—	/v1/*

Semantics:

• Unset means shared, so upgrades are a no-op and existing callers keep working.
• api.port must differ from http.port. Setting them equal is rejected at config load, rather than failing later as an opaque bind error. Omit the api block to share the port.
• Callers of /v1/snapshot and friends must move to the new port once it is set.

There is still no authentication on /v1. This narrows the blast radius by network topology only, so keep the API port network-internal. In the Helm chart that means a second containerPort, and the /v1 port must not be added to the PodMonitor.

Consumer Development

All consumers are compiled into the single bin/siphon binary; the consumer config's type selects the implementation. Adding a consumer type:

• pkg/config/consumer.go
  • Add a ConsumerType constant and, if the type needs settings, a typed config block on ConsumerConfig (validated in ConsumerConfig.Validate).
• pkg/consumers/xyz_consumer.go
  • A constructor matching consumers.BuildHandler: func NewXyzConsumer(ctx context.Context, cfg *config.ConsumerConfig, logger *zap.SugaredLogger) (EventHandler, error).
  • The handler implements consumers.EventHandler:
    • HandleEvents(ctx context.Context, pkg *pb.LogicalReplicationEvents, stream *config.Stream) error handles one package from NATS. pkg holds the logical replication events, and stream is the stream configuration (it tells the handler where to write the data). Return an error when the package must be retried; the message is then NAKed.
    • Close() error releases the handler's resources; return nil if there are none.
  • Tests go next to it in pkg/consumers.
• cmd/main.go
  • Add a case for the new type in the consumer dispatch.

See the demo consumer for a minimal example.

The LogicalReplicationEvents Package

The LogicalReplicationEvents package contains logical replication event data for a given PostgreSQL database table or partition.

The following attributes are available:

• event: Type of the event package: CDC_EVENT (1, ordinary CDC and snapshot rows), CLICKHOUSE_REFRESH_EVENT (2), REFERENCED_CDC_EVENT (3, the payload is in object storage; consumers resolve it before the handler sees it), HEARTBEAT (4, no events) or CORRECTION_EVENT (5, a consistency correction).
• table: The PostgreSQL table name associated with the event. For partitioned tables, this will be the name of the partition.
• schema: The schema (namespace) of the PostgreSQL table.
• version_hash: Hashed database schema (table name, column names and their types).
• application_identifier: Identifies which Siphon producer processed the event. Useful as a surrogate key when packages from different PostgreSQL databases are sent to the same stream.
• events: A array of logical replication events.
• columns: A list of column names used to resolve the column_index field in each item of the events array.
• correction_filters: For a CORRECTION_EVENT, the predicates describing exactly which rows were snapshotted, so a consumer can clear the same set before inserting.
• column_type_oids: The PostgreSQL type OIDs (pg_type.oid) of the columns, index-aligned with columns. May be empty: ClickHouse refresh packages carry no PostgreSQL relation, and older producers do not send the field at all. Only use it when column_type_oids and columns have the same length. A 0 OID is a special case meaning the type was not available for that column, which should not happen under normal circumstances.

Package Guarantees

• Events within a package are ordered identically to their appearance in the PostgreSQL WAL log.
• All events in a package conform to the same table schema. Note: DELETE events only contain primary key values.

Large Packages

When an event package exceeds the configured payload limit, Siphon automatically diverts the data to a configurable object storage backend to prevent "payload too large" errors. Snapshot events are also unconditionally stored in object storage regardless of size. A lightweight reference message is published to the NATS stream instead; the consumer transparently fetches the full event from object storage before processing.

The object storage backend is configured via queueing.object_storage_config:

queueing:
  object_storage_config:
    identifier: "my-object-storage"
    type: "gcs"              # "gcs", "s3", "file", or "nats"
    bucket_name: "my-bucket" # GCS or S3 bucket, local directory, or NATS ObjectStore bucket

When object_storage_config is omitted entirely, it defaults to NATS ObjectStore with bucket repl_events_bucket -- preserving backward compatibility with existing deployments. When provided, all three fields (identifier, type, bucket_name) are required.

Supported backends:

• gcs: Google Cloud Storage. Recommended for production.
• s3: Amazon S3 or an S3-compatible server. Credentials and region come from the AWS SDK default chain; set AWS_ENDPOINT_URL_S3 for MinIO or LocalStack.
• file: Local filesystem. Useful for development and testing.
• nats: NATS ObjectStore (the default). Uses the existing NATS connection (no extra connection created).

Objects are stored under {application_identifier}/cdc/{stream_name}/{subject}/{random_suffix} within the configured bucket, so multiple producers can safely share a bucket.

You can fine-tune the oversize threshold:

oversize_limit_bytes: Manually defines the threshold size at which events are moved to object storage. If left undefined, Siphon defaults to the maximum payload size configured in your NATS server.

Emergency Mode (NATS unavailable)

When NATS is reachable, Siphon's CDC pipeline is the standard producer → NATS → consumer flow. If NATS becomes unreachable, the producer would normally crash-loop, blocking PostgreSQL's WAL slot from advancing and growing replication lag. Emergency mode prevents that: when configured, the producer keeps running with NATS down, buffering compressed-protobuf CDC events to a durable object-storage backend (GCS, S3 or local file). A monitor goroutine probes NATS in the background; when it recovers, the buffered events are drained to NATS in LSN order and normal replication resumes.

While emergency mode is active:

• WAL slot still advances. SendStatusUpdate to PostgreSQL continues, so confirmed_flush_lsn keeps moving and PostgreSQL doesn't accumulate WAL.
• Snapshotting is disabled. New tables added during the outage are not snapshotted and the publication is not modified — those operations require NATS. They are reconciled on the next normal-mode boot.
• Partition monitoring is disabled. Detection still works (it queries PG), but acting on a partition change requires SetupPublication + an initial snapshot, both of which need NATS. Letting the monitor fire would just restart the producer back into emergency mode without actually adding the new partition. Partitions added during the outage are picked up on the next normal-mode boot.
• At-least-once delivery is preserved. Each emergency object is named with the LSN as a 16-char zero-padded hex prefix, so listing returns events in LSN order on every backend. On drain, each object is published to NATS and only deleted after the publish ACKs.

Configure under queueing.failover_storage:

queueing:
  driver: nats
  url: "nats://my-nats:4222"
  stream_name: my-stream
  object_storage_config:
    identifier: gcs-object-storage
    type: gcs
    bucket_name: my-siphon-bucket      # oversize/snapshot events stored here
  failover_storage:
    storage:
      identifier: gcs-emergency        # any non-empty string; logged + stamped on objects
      type: gcs                        # "gcs", "s3" or "file"
      bucket_name: my-siphon-bucket    # GCS/S3 bucket or local directory (file backend); can share the same bucket as object_storage_config
    recovery_poll_seconds: 30          # how often to probe NATS for recovery (default 30)
    max_concurrent_drain_subjects: 4   # parallel drain goroutines (default 4)

The object_storage_config and failover_storage can safely share the same bucket -- they use different path prefixes ({appID}/cdc/... vs {appID}/{stream}/...) to avoid collisions.

For local development / testing, swap GCS for the file backend:

queueing:
  object_storage_config:
    identifier: file-object-storage
    type: file
    bucket_name: /var/lib/siphon/objects
  failover_storage:
    storage:
      identifier: file-emergency
      type: file
      bucket_name: /var/lib/siphon/emergency

When failover_storage.storage is unset, emergency mode is disabled and pre-existing crash-loop behavior is unchanged.

Multi-tenant safety: every emergency object is written under {application_identifier}/{stream_name}/{key_path}/{lsn_hex16}_{rand} in the configured bucket. Multiple producers can share a bucket without colliding as long as their application_identifiers differ — there's no per-storage prefix or base-path config to manage.

Startup safety: at boot, the producer validates the failover_storage config (type, identifier, bucket) and runs a write/read/delete probe against the backend. A misconfigured bucket or missing credentials crashes the producer with a clear error instead of silently failing the first time emergency mode tries to engage.

NATS health check: the recovery monitor doesn't accept "TCP handshake works" as healthy. A NATS cluster with a stuck metadata leader or with the main stream / oversize bucket missing would let the producer exit emergency mode and immediately thrash on Publish failures. The check requires the main stream to be reachable via Info, and, when object_storage_config uses the NATS backend, the oversize ObjectStore bucket too — only then does the producer transition out of emergency mode. GCS, S3 and file object storage are independent of NATS health and are not probed.

Metrics:

• siphon_failover_mode_active{app_id} — gauge, 1 while emergency mode is active.
• siphon_failover_writes_total{app_id, schema, table} — events buffered to durable storage.
• siphon_failover_drain_published_total{app_id} — events drained back into NATS on recovery.
• siphon_failover_drain_errors_total{app_id} — drain-loop errors (a run will be retried on the next iteration).

Example: Partitioned Table

Producer Configuration:

table_mapping:
  - table: audit_events
    schema: public
    subject: audit_events

Consumer Configuration:

streams:
  - identifier: audit_events
    subject: audit_events
    target: siphon_audit_events

In this example, the audit_events table is partitioned. The Siphon producer automatically expands it to include all its partitions. On the consumer side, each event package contains events from only one partition.

If 2 records are inserted into the public.audit_events_2023 partition and 5 records into the public.audit_events_2024 partition, the Siphon consumer will receive 2 event packages:

• public.audit_events_2023: 2 events
• public.audit_events_2024: 5 events

Releasing

Note: This is for project maintainers or above

To release a new version of Siphon:

• Ensure main contains the latest changes to be released.
• Create a new git tag. git tag <tagname>
  • tagname should be a semantic version number.
  • Push the tag to the repository. git push --tags.
• This kicks off a release pipeline that will:
  • Run the entire test suite.
  • Build the Docker file.
  • Push it to the Container Registry tagged as both the defined version, and latest.
  • Build and push the FIPS variant as <version>-fips and latest-fips (see below).
  • Sign both images with Sigstore/cosign (see Image signing).
  • Create a new release in GitLab.

FIPS images

Every release also publishes a FIPS 140-3 variant under a -fips suffix:

Tag	Crypto
<version>, latest	Go standard library, FIPS mode off
<version>-fips, latest-fips	Go Cryptographic Module v1.0.0 (CMVP certificate #5247), FIPS mode on

Both are built from the same source and the same Dockerfile; the FIPS variant just passes --build-arg GOFIPS140=v1.0.0. It stays CGO_ENABLED=0 on the same distroless base, so there's no OpenSSL to link and multi-arch builds still cross-compile without QEMU.

FIPS mode is baked into the binary as its default GODEBUG, so there's nothing to set at deploy time. To confirm what's running, check the crypto line the process logs at startup:

{"level":"info","msg":"crypto","fips140":true,"module":"v1.0.0"}

Build one locally with:

docker buildx build --build-arg GOFIPS140=v1.0.0 -t siphon:fips .

Two things worth knowing:

• FIPS mode narrows the TLS versions, cipher suites and curves Go will negotiate. If ClickHouse, NATS or object storage is pinned to something outside that set, the -fips image won't connect where the regular one does.
• NATS nkeys use ed25519, which isn't FIPS-approved. That's fine under fips140=on, which still permits non-approved algorithms. Don't run with GODEBUG=fips140=only. It's a debugging aid that turns those calls into panics, not a supported deployment mode.

Image signing

Released images — both the regular and the -fips variant — are signed with Sigstore keyless signing. Signatures are stored alongside the image in the container registry and the certificate is recorded in the public Rekor transparency log.

Only tagged releases are signed.

Verify an image before running it. Releases are signed with cosign 3.x, and cosign 2.x cannot read v3 signatures, so verifiers need cosign >= 3.0.0:

cosign verify \
  registry.gitlab.com/gitlab-org/analytics-section/siphon:<version> \
  --certificate-identity "https://gitlab.com/gitlab-org/analytics-section/siphon//.gitlab-ci.yml@refs/tags/<version>" \
  --certificate-oidc-issuer "https://gitlab.com" | jq .

To verify the FIPS variant, add the -fips suffix to the image reference only — the certificate identity stays the same:

cosign verify \
  registry.gitlab.com/gitlab-org/analytics-section/siphon:<version>-fips \
  --certificate-identity "https://gitlab.com/gitlab-org/analytics-section/siphon//.gitlab-ci.yml@refs/tags/<version>" \
  --certificate-oidc-issuer "https://gitlab.com" | jq .

Both images are built and signed by the same release_job run off the same git tag, so their Fulcio certificates carry an identical workflow ref.

The double slash before .gitlab-ci.yml is required.

Signatures carry annotations recording the pipeline that produced them:

cosign verify ... | jq '.[0].optional'

which reports com.gitlab.ci.pipeline.id, com.gitlab.ci.user.name, the tag that was pushed, and variant=fips on the FIPS image.

Signing targets the image digest rather than the tag, so a signature stays valid and unambiguous even after latest and latest-fips move to the next release.

Configuration and Secret Management

The unified config.yml often requires sensitive information, such as database passwords. To secure these values, Siphon's built-in YAML parser resolves environment variables, eliminating the need to store secrets directly inside the file.

Example:

#...
clickhouse:
  host: localhost
  port: 9000
  username: default
  password: ${CH_DB_PASSWORD}
  database: gitlab_clickhouse_development
#...

CH_DB_PASSWORD=password bin/siphon --config config.yml

Limitations

Siphon uses the default pgoutput logical replication plugin, processing the data has some limitations:

• Computed/Generated Columns: Not replicated (GENERATED ALWAYS AS).
• Historical BC Dates: Very old dates/timestamps in BC will not be parsed correctly; the BC suffix is ignored. The consistency checker reproduces the same trim so BC rows do not read as divergence, but a BC year reinterpreted as AD outside DateTime64's [1900, 2299] cannot be stored by the target and diverges regardless. Workaround: enable replication.preserve_postgresql_timestamp_representation_as_string: true to store timestamps as raw PostgreSQL strings (see below).
• infinity Dates and Timestamps: infinity and -infinity in date, timestamp, and timestamptz columns are replicated as the bounds of google.protobuf.Timestamp: 9999-12-31 23:59:59.999999 (9999-12-31 for a date) and 0001-01-01 00:00:00 UTC. In ClickHouse DateTime64 columns they land as unrelated dates, because the ClickHouse Go driver encodes a time as nanoseconds since the epoch, which overflows outside roughly [1678, 2262]. This applies to any value outside that range, such as a literal 9999-12-31.
• Timezone Normalization: All timestamps are converted to UTC. For timestamps without time zone, the PostgreSQL server's timezone is used as the base. Workaround: enable replication.preserve_postgresql_timestamp_representation_as_string: true to preserve the original timezone offset (see below). Note: the consistency checker hashes a timestamp without time zone as UTC, which is the wire contract, so on a server whose timezone is not UTC a custom-mode check covering such a column reports divergence that is not there. Run the source in UTC, or leave those columns out of the list.
• Oversized Column Values: A column value above max_column_size_in_bytes is replicated as an empty string. A custom-mode consistency check covering such a column reports it as divergent, because the source hashes the full value.
• Array NULL Elements and Dimensions: The protobuf array types cannot express a NULL element or a multi-dimensional array. Workaround: enable replication.preserve_postgresql_array_representation_as_string: true (see below).
• Nested Partitions: Siphon only discovers direct child partitions. Multi-level partitioning hierarchies are not automatically expanded.
• Sequential Snapshotting: Initial data snapshots are processed sequentially (one table at a time).
• CDC History Gaps for New Partitions: PostgreSQL only generates logical replication messages for tables that are members of a publication. There is a technical window between a partition's creation and its detection by Siphon where individual row changes (DML) will not be captured in the live CDC stream, although the data will be included in the subsequent initial snapshot (eventually consistent).

Preserving PostgreSQL Timestamp Representation

When replicating to a target that accepts PostgreSQL-native timestamp strings (e.g. another PostgreSQL instance), you can enable replication.preserve_postgresql_timestamp_representation_as_string: true in the producer config. When enabled:

• date, time, timetz, timestamp, and timestamptz column values are stored as raw PostgreSQL string literals (e.g. "4714-11-24 00:00:00+00 BC") instead of being parsed into Go time.Time values.
• BC-era dates and original timezone offsets are preserved exactly as PostgreSQL sends them.
• The protobuf StringValue field is used for these columns instead of TimestampValue.

Note: This option is not recommended for ClickHouse consumers, which expect TimestampValue for DateTime64 columns.

Preserving PostgreSQL Array Representation

When replicating to another PostgreSQL instance, enable replication.preserve_postgresql_array_representation_as_string: true. Array columns are sent as the raw literal in the protobuf StringValue field, and the target's array_in() parses them back.

Note: not recommended for ClickHouse consumers, which expect Array(T).

PostgreSQL Consumer

• DELETEs always flow through. row_filters only restrict INSERT and UPDATE events. Every DELETE from the source is forwarded directly to the target — there is no producer-side or consumer-side suppression of DELETE events. The design relies on disjoint primary-key ranges between source and target: a DELETE WHERE pk = X for a row that does not exist on the target is a silent no-op, so cross-cell DELETEs are harmless. If source and target can share primary keys, an in-scope source DELETE may remove a target row that legitimately belongs to a different cell. Operators that cannot guarantee disjoint ID ranges should not use the PG sink with row_filters.

Reporting Bugs and Issues

Where to report

• Issues, bugs, and questions: For GitLab-related Siphon issues or Siphon infrastructure problems, open an issue in this project and mention the @gitlab-org/analytics-section/siphon group.
• Urgent or synchronous matters: Reach out in the #f_siphon Slack channel.

What to include

To help us investigate quickly, please include the following in your report:

1. Siphon version: The Docker image tag or the Git commit SHA of the Siphon build you are running.
2. Logs: Logs from the affected component (producer, consumer, reconciler, etc.) covering the time of the problem. Debug-level logs are the most useful when available; they can be enabled with the --log-level debug flag.
3. PostgreSQL topology: The deployment setup in use (single database, replica, dedicated snapshot database, multi-producer; see Deployment Setup) and the PostgreSQL version.
4. Configuration: The Siphon configuration used to run the affected component(s), without secrets or sensitive data.
5. Dependency details: Versions of the related services where relevant, for example NATS and the target data store (ClickHouse, Snowflake, etc.).

Note: Before sharing logs or configuration, redact any secrets and sensitive data such as passwords, tokens, connection strings, and customer data.