S3 sink

queen-s3 is a separate binary that reads a queue’s log through Queen’s own HTTP API and writes it into an object store in an open format. It is a client of Queen like any SDK is: it holds no database connection, embeds no broker, and stores nothing durable of its own except three small documents per queue in Queen’s key/value store (s3:<sink>:<queue>:{intent,committed,lease}) and what it puts in the bucket. What it writes, and what a reader gets, is the lake reference.

It is not a wire-protocol facade. Nothing connects to it, and it advertises no address. The two ports it opens are /healthz and /metrics, on loopback by default.

The two shapes

Beside a broker (self-hosted). The sink runs anywhere that can reach a broker, points QUEEN_URL at it, and carries one QUEEN_TOKEN. It reads through POST /api/v1/fetch and POST /api/v1/partitions/changed, and keeps its commit pointers in POST /api/v1/kv. Nothing else on the broker changes: no push path, no pop path, no retention, no table.

Beside a proxy (Cloud). The same binary with QUEEN_URL pointed at a cell’s proxy. Every call then crosses the proxy’s authentication, tenant scoping, quotas and metering exactly as an SDK’s call does, and the sink is an ordinary tenant process reading its own queues. See In Queen Cloud for the two routes and the key/value carve-out that make that work.

Embedded in the broker. QUEEN_S3_EMBEDDED=true makes the broker spawn and supervise the sink as a child process, wired to the broker’s own listener over loopback. One deployment, two processes, one image. That is its own section.

The choice between the first and the third is not about capability, it is about where the process lives. A sink is a client and needs nothing from the broker’s container, so running it separately is what lets it be scaled, restarted and upgraded on its own schedule. Embedded mode exists so a self-hoster does not have to operate a second thing.

Build and run

The crate is standalone, so it builds on its own manifest and produces one binary.

cargo build --release --manifest-path connectors/queen-s3/Cargo.toml
QUEEN_URL=http://localhost:6632 \
QUEEN_S3_QUEUES=orders \
QUEEN_S3_ENDPOINT=https://s3.eu-central-1.amazonaws.com \
QUEEN_S3_REGION=eu-central-1 \
QUEEN_S3_BUCKET=my-lake \
QUEEN_S3_ACCESS_KEY=AKIAEXAMPLE \
QUEEN_S3_SECRET_KEY=a-secret-of-your-own \
  ./connectors/queen-s3/target/release/queen-s3

Two rules hold for every variable below. A value that does not parse is a fatal boot and never a silent fall back to the default, because the default is not there to paper over a typo. And a variable that is present but empty or whitespace-only counts as unset, because an empty value in a Compose file or a Helm template is a variable somebody meant to set.

Boot, and the four exit codes

Boot is short and it is all before the first byte is read: the configuration, the health listener, one client for Queen and one for S3, then two probes, both before any queue is claimed. The first is an empty POST /api/v1/partitions/changed, which answers safeTime alone, writes nothing, and is classed Consume at the proxy exactly like the fetch and the commit, so it proves QUEEN_URL and QUEEN_TOKEN with the scope the sink needs anyway (see below); the second is a HEAD of the prefix, where a 404 is a fine answer (nothing has been written yet) and a 403 or a connection failure is the credential or the endpoint being wrong. Both failures print what to check, not just the error.

Code 3 is deliberately separate from the running failure policy, which is to lag rather than to die: once the sink is up, an unreachable broker or bucket is a retry behind a backoff and a red /healthz. At boot the same symptom is almost always a wrong URL, a wrong key or a missing network, and a container that exits says so louder than one that sits there logging.

Then the process is a task per queue, and main owns only three things around them. The queue set: a named list is fixed for the life of the process, while QUEEN_S3_QUEUES=* is re-listed every ten discovery intervals (never faster than once a second), so a queue created at ten in the morning is being mirrored a minute later. Queues are never removed from the set: a task whose queue disappeared meets UNKNOWN_TOPIC_OR_PARTITION, stops, and retries once a minute, which is also what should happen when the queue comes back. Ownership: one lease per queue, retried after its TTL when another instance holds it. The signals: SIGTERM and Ctrl-C stop new reads, let the window in flight finish, and exit 0. That is the same drain rule the embedded supervisor’s grace is sized for, and it is described in full below.

Every variable

The four with no default are the destination: QUEEN_S3_ENDPOINT, QUEEN_S3_REGION, QUEEN_S3_BUCKET and QUEEN_S3_QUEUES, plus the keypair. There is no bucket that is right more often than it is wrong, so there is no default for one.

QUEEN_S3_BIN, QUEEN_S3_EMBEDDED and QUEEN_S3_SHUTDOWN_GRACE_MS are read by the broker, not by this process, and are in the table below.

The API key and its scopes

The sink makes three or four kinds of call, and the credential has to cover all of them. Against a proxy the answer is short: consume, and consume alone, unless the queue list is *.

The last row is the one that catches people. A named queue list never touches the queue listing, so a consume-only key runs a named sink end to end. * resolves through GET /api/v1/resources/queues, which the proxy classifies as a read route, and for an API key Read means read or admin: a consume-only key is answered 403 there and the process exits 3 at boot with the “cannot reach Queen” line, before it has claimed a queue or touched the bucket. So * costs a token the read scope on top of consume, and the failure is nowhere near the bucket. Against a broker rather than a proxy, read-only already covers the same call.

Against a broker with JWT_ENABLED=true the levels are the broker’s own rather than the proxy’s, and the binding one is the key/value write: POST /api/v1/kv is read-write, while both reads are read-only. A read-write token therefore covers the sink and a read-only token covers only two thirds of it, which is a sink that reads a whole window and can never commit it.

With authentication off the broker verifies nothing and QUEEN_TOKEN is an identity rather than a credential.

The S3 side

Any S3 API works: AWS, DigitalOcean Spaces, Cloudflare R2, and versitygw, which is what the tests run against. The sink deliberately uses no conditional PUT, no LIST on the correctness path and no read-after-write assumption, because a retried upload of a window is byte-identical and overwriting a key with the same bytes is the whole idempotency story. That is what makes it work on a gateway whose LIST is eventually consistent and on a service that has no conditional PUT to offer.

Integrity is checked both ways the protocol offers. A single PUT sends Content-MD5 and compares the answer’s ETag against the MD5 it computed; a completed multipart upload’s ETag carries a -<parts> suffix, and the part count is compared with the number of parts that were sent. Two shapes are not the ones integrity is defined over, and both are reported at debug rather than failed: an ETag under SSE-KMS is opaque by an AWS rule, and some gateways answer a multipart ETag with no suffix at all.

Set an encryption policy on the bucket. QUEEN_S3_SSE=AES256 puts the header on every PUT, and aws:kms with QUEEN_S3_SSE_KMS_KEY_ID makes the KMS key policy the real access control. Say plainly what that is protecting: the bucket holds a decrypted copy of the log. A fetch decrypts a payload when the broker has an encryption key configured, so a queue that is encrypted at rest inside Queen is plaintext in the lake, and at-rest encryption on the bucket is what restores the property rather than an extra.

The bucket policy, least privilege

Five verbs and no more, and the fifth is narrower than it looks: the sink never deletes a data object, only its own checkpoints.

{
  "Version": "2012-10-17",
  "Statement": [
    { "Sid": "ListPrefix",
      "Effect": "Allow",
      "Action": ["s3:ListBucket"],
      "Resource": "arn:aws:s3:::my-lake",
      "Condition": { "StringLike": { "s3:prefix": ["queen/*"] } } },
    { "Sid": "WriteAndRead",
      "Effect": "Allow",
      "Action": ["s3:PutObject", "s3:GetObject", "s3:AbortMultipartUpload"],
      "Resource": "arn:aws:s3:::my-lake/queen/*" },
    { "Sid": "GarbageCollectCheckpointsOnly",
      "Effect": "Allow",
      "Action": ["s3:DeleteObject"],
      "Resource": "arn:aws:s3:::my-lake/queen/_queen/*/checkpoint/*" }
  ]
}

queen/ is QUEEN_S3_PREFIX. Add an AbortIncompleteMultipartUpload lifecycle rule while you are there: the sink aborts its own multipart uploads, but a process killed between two parts leaves one behind, and a lifecycle rule is what collects those without anybody watching.

Grant pg_read_all_stats to the broker role

This is the one requirement that lives on the database rather than on the sink, and it is worth doing before the first window closes.

A window may only close at or below safeTime, a watermark the broker computes as the oldest xact_start among in-transaction sessions: below it, nothing new can still become visible, which is what makes a window a deterministic set. PostgreSQL masks pg_stat_activity across roles. For a session owned by a role the reader is not a member of, and without pg_read_all_stats, state and xact_start come back NULL, and a masked row contributes nothing to a minimum while being exactly the kind of session that could be holding a write transaction open. Taking the minimum over the visible rows would overestimate the safe instant, which is the one direction that loses data.

So masked rows are counted, and one is enough: the broker falls back to now() - QUEEN_FETCH_SAFE_FLOOR_MS (default 30 seconds) and answers safeTimeDegraded: true.

Measured, not predicted: on PostgreSQL 16 backend_type is masked too, and PostgreSQL’s own background workers belong to the bootstrap superuser, so for any non-superuser broker role without pg_read_all_stats the degrade is permanent rather than occasional.

GRANT pg_read_all_stats TO queen;

Degraded is usable. It costs lag and nothing else: every window closes 30 seconds behind instead of five, the sink logs the state once and never stalls on it, and no record is lost or duplicated either way. The grant is what buys the 25 seconds back.

The lag SLO, and the retention rule

queen_s3_lag_seconds{queue} is the number to alarm on. It is now minus the end of the last committed window, so in a healthy sink it oscillates below QUEEN_S3_MAX_WINDOW_MS plus the safe lag: about five and a half minutes at the defaults, or six with a degraded safeTime. Alarm at a multiple of that, not at the number itself.

It is worth alarming on because of what sits behind it. If retention deletes a segment before the sink read it, that is the one failure that is data loss, and the lag gauge is the thing that was rising for hours beforehand. The loss itself is never silent: the offset gap goes into the window’s manifest as lost, counts on queen_s3_records_lost_total{queue}, gets one sampled log line, and the sink resumes from the new log start and keeps committing, because a stalled sink loses more than a lagging one.

The rule that prevents it: a queue’s retentionSeconds must comfortably exceed QUEEN_S3_MAX_WINDOW_MS plus the longest outage the sink is expected to survive. Five minutes of window against a day of retention is not a decision anybody has to think about; five minutes of window against ten minutes of retention is one bad deploy away from a hole.

Or make it structural. A queue can name the sink that has to keep up with it, in two options on POST /api/v1/configure:

The floor retention applies is GREATEST(committed window end - 60 s, now() - retentionSinkHoldMaxSeconds), and both halves earn their place. The minute of slack covers the sub-second skew of a watermark-derived bound, paid in kept bytes and never in lost ones. The cap is what made the option shippable at all: without it a stopped sink is unbounded retention and the first symptom is a full disk. It also means the hold is safe from the moment it is set, before the sink has ever committed, because a queue with no pointer yet keeps everything younger than the cap.

Two limits worth knowing before you rely on it. The hold floors the two cutoffs that delete segments, and deliberately not the dedup-hash purge or the max-wait eviction: those delete by different rules, and a copy in a lake says nothing about when a dedup hash may go or about a message too old to be worth delivering. And the pointer read degrades rather than fails: if it cannot be read, every held queue falls back to its cap-only floor, because that value is tenant-writable and a malformed one must never be able to stop deletion cluster-wide.

Embedded in the broker

QUEEN_S3_EMBEDDED=true makes the broker spawn and supervise the sink as a child process, wired to the broker’s own listener over loopback and derived from the address that listener actually bound, so PORT=0 and a wildcard bind both come out right.

It is a child and not a library on purpose. The sink has its own accept loop, its own compression and Parquet arenas, its own S3 client and its own crash modes, and the broker is built with panic = "abort", so an allocation failure in a writer sized from QUEEN_S3_TARGET_MB would take the broker down rather than one task. A child keeps the blast radius at the sink: it dies, the broker keeps serving, and the supervisor brings it back on a backoff that doubles from one second to a thirty second ceiling and resets after an hour of healthy running. A restart costs lag and never a gap, because the sink resumes from its committed pointer.

These three are read by the broker, not by the sink:

Every other QUEEN_S3_* variable means exactly what it means when the sink runs alone. The environment forwards verbatim, QUEEN_TOKEN and the S3 keypair included, because those are the child’s own credentials for the two hops it makes. Four of the broker’s secrets are stripped before exec and the sink reads none of them: PG_PASSWORD, JWT_SECRET, QUEEN_ENCRYPTION_KEY and QUEEN_SYNC_SECRET. A database password has no business being readable in a second process’s environment just because that process happens to be colocated.

Three mistakes are refused at boot rather than discovered as a crash loop: a QUEEN_S3_BIN that is not a file, a binary that exists but has lost its execute bit, and QUEEN_S3_EMBEDDED=true with no QUEEN_S3_BUCKET. The third names all four required variables in its message, because nobody arrives at a bucket name without an endpoint, so the fix is one edit. A fourth is a warning and not a refusal: JWT_ENABLED=true with no QUEEN_TOKEN, where every call the child makes is answered 401 and nothing is ever written. The loopback hop is authenticated like any other client’s, and embedded mode gets no private door into the broker, because a private door is exactly the kind of thing that is later found open from somewhere else.

The loopback is the default, not the only answer. An explicitly set, non-empty QUEEN_URL in the broker’s own environment wins, and the child is given that instead, which is what makes embedded mode usable in a cell where the sink has to reach the broker through the proxy. The boot line says which branch was taken:

queen-s3 sink started (embedded) pid=18847 bin=/app/bin/queen-s3
  queen_url=http://queen-proxy:6711 queen_url_from="QUEEN_URL (explicit)"

queen_url_from is either QUEEN_URL (explicit) or loopback (bound listener).

The child’s stdout and stderr are forwarded into the broker’s log, stripped of escape sequences and control bytes, truncated at 4 KiB a line, and budgeted at 200 lines per ten second window with a suppressed count when the budget is spent. GET /status grows an s3 block when embedded mode is on, and only then: mode, phase, pid, restarts, lastExit, uptimeMs and backoffMs.

What a stop guarantees, and what it does not

The child is put in its own process group, so a stop signals the whole group and a grandchild dies with it rather than being re-parented.

• The broker gets SIGTERM or Ctrl-C: the supervisor runs after the broker’s serve loop drains, sends SIGTERM to the group, waits up to QUEEN_S3_SHUTDOWN_GRACE_MS, and escalates to SIGKILL. A broker stopping with a sink attached therefore takes up to half a minute longer than one stopping with a facade. That is not overhead, it is the window being finished.
• The broker panics or drops the handle: the child is reaped with it.
• The broker is itself SIGKILLed: on Linux the child carries PR_SET_PDEATHSIG, so the kernel kills it whatever the parent died of. On macOS and BSD there is no equivalent: a SIGKILLed broker leaves the sink running, re-parented to init and still writing to the bucket. A development machine caveat rather than a production one, stated rather than papered over.

What the grace buys is bounded and worth being exact about. On SIGTERM the sink stops reading and decides once, per queue, whether the window it is filling is worth finishing: it is if at least 1 MiB is buffered, or if the window has been open at least ten seconds on the broker’s clock and holds anything at all. A window worth finishing is closed, uploaded and committed inside the grace. Then: either that commit lands, or the next start redoes the window from its intent and writes the identical object under the identical key. Nothing is lost in the second case; the work is repeated. Cutting the grace to a facade’s five seconds does not risk the lake, it just makes the repeat more likely.

The image

The repository’s Dockerfile builds the sink in a stage of its own and copies it next to the broker binary in /app/bin. That adjacency is the contract: with QUEEN_S3_EMBEDDED=true and no QUEEN_S3_BIN, the supervisor resolves the child out of its own executable’s directory, so embedded mode needs no extra configuration in that image. The same image runs the sink alone instead, with docker run … queen-mq ./bin/queen-s3, which is the shape that scales out, because a sink needs nothing from the broker’s container.

The connector has a CI job from its first commit, running its unit tests, rustfmt and clippy on its own manifest. That is deliberate and it is a correction: the SQS facade shipped in 1.4.0 with 715 tests and no job at all, so for a release those tests only ever ran where somebody happened to run them. A suite CI never executes is a suite whose green is a memory.

Scaling out

Windows are per queue, so one queue’s throughput is one instance’s. Scaling out means splitting the queue list across processes, not adding replicas over the same one.

Instances that do overlap are safe and wasteful rather than dangerous. A queue is owned through a lease in Queen’s key/value store, refreshed on a tick and expiring after QUEEN_S3_LEASE_TTL_MS, and every intent and commit carries that lease as a required precondition. Two instances racing for one queue means the loser’s compare-and-set fails, its batch rolls back whole, and queen_s3_commit_precondition_lost_total counts it. Two instances can never commit two different versions of the same window.

The practical consequence for Kubernetes: enabling the sink on a StatefulSet with more than one replica gives every pod the same queue list. Run it on a single-replica release, or split the queues across releases.

/healthz and /metrics

Both are served on QUEEN_S3_LISTEN, which defaults to loopback.

/healthz answers one question, and it is the failure policy rather than a subsystem tree: the sink never drops, it only lags. It is 200 while every queue this process owns has committed a window inside 3 × QUEEN_S3_MAX_WINDOW_MS (floored at 30 seconds), and 503 when one has not, naming the first offender by name so the line is stable across scrapes:

{"ok":false,"queue":"orders","staleMs":1200000,"limitMs":900000}

Three windows and not one: a window closes at MAX_WINDOW_MS, then has to be uploaded and committed, and the broker’s own safeTime lags by its guard, so a one-window threshold would go red on a healthy sink under load, and a probe that flaps is a probe an operator turns off. A queue that has never committed is green, which is the honest answer for a lane nobody has pushed to yet and for the seconds after boot.

An unreachable broker, a bucket refusing every PUT, and a lease lost to another instance all arrive here as the same symptom, because all three have the same consequence.

/metrics is Prometheus text exposition, version 0.0.4:

The two histograms are deliberately unlabelled: a histogram per queue is buckets times queues in series, and the per-queue answer worth having is the lag gauge rather than the size distribution.

On Kubernetes

The repository ships no Helm chart, so a Kubernetes deployment is a manifest an operator writes. Four things belong in it, and none of them is obvious from the variable table alone.

• The S3 keypair comes from a Secret, always. It is the credential for the hop the broker never makes, and it does not belong in a values file or a plain env: block.
• QUEEN_S3_LISTEN has to be widened to be scraped. The default is loopback, which is right for a sidecar and wrong for a ServiceMonitor. Set 0.0.0.0:9333 and add the containerPort.
• One replica, for the reason under scaling out: a second replica gets the same queue list and races the first for every window.
• The readiness probe is /healthz, and its budget is three windows, so a MAX_WINDOW_MS of five minutes wants a probe timeout comfortably inside fifteen minutes rather than one that restarts the pod during a slow window.

In embedded mode there is nothing extra to deploy: the three broker variables above go on the broker’s own workload, and the sink is a process inside the pod that already exists.

In Queen Cloud

Queen Cloud runs the broker as a managed cell, and has a free tier.

In Cloud the sink does not talk to a broker, it talks to the cell proxy, and the proxy talks to the broker. That is the whole of the difference, and it is one variable: point QUEEN_URL at the proxy’s Service. Every call then crosses the proxy’s authentication, tenant scoping, quotas and metering exactly as an SDK’s call does.

Both routes the sink reads through are Consume, and neither is ever quota-blocked. POST /api/v1/fetch and POST /api/v1/partitions/changed are classified for the authority they need rather than for what they write. Neither writes anything, but one hands out message payloads and the other hands out the partition names, offsets and retention watermarks to read them by, so both carry the authority of the pop they stand beside instead of the read level every user role already has. A blocked read would be worse here than for a consumer: a sink refused discovery cannot even learn which partitions it is behind on, so it stops mirroring a queue whose backlog keeps growing, which is the opposite of what a storage block is for. Both are POST on the exact path only, so every other spelling fails closed.

The commit pointers are reclassified. POST /api/v1/kv is normally gated on the kv feature and answers the storage block on writes. A batch that addresses nothing but the sink’s reserved space, namespace queen-s3 and key prefix s3:, is reclassified Consume at the gateway, which has two consequences worth stating separately. A tenant whose plan has never heard of the kv feature can still run a sink. And a tenant over its storage quota can still commit a window, which is the trap this exists for: refusing a commit pointer of a few hundred bytes does not stop the tenant growing, it makes the sink re-upload and re-commit the identical window for ever while its lag grows without bound.

Everything about that carve-out fails closed. An unreadable body, an empty batch, an operation the proxy has not been told about, one foreign key anywhere in the array: all of them mean “not a sink batch”, which means today’s gating. The cost of getting it wrong in that direction is a stalled sink with a loud lag gauge; the cost in the other direction would be a tenant slipping the feature gate with an arbitrary key.

Nothing here is message metered. A fetch and a discovery call each book one request and zero messages, exactly as ack and lease extension do. A tenant mirroring a million records into its own bucket is billed for the requests that carried them, not for the records, and the egress to the object store is the tenant’s own bill with its own provider. That is a pricing decision rather than a gap, and it is stated here so nobody discovers it from an invoice.

The tenant of every call is the proxy’s, from its trusted header; the sink sends only its token.

What the boot log says

One line, and it is the one to compare against what the bucket actually contains:

queen-s3 1.5.0 sink=default instance=sink-0 queues=orders,clicks partitions=discovery
  queen=http://localhost:6632 token=<set> s3=https://s3.eu-central-1.amazonaws.com
  bucket=my-lake prefix=queen region=eu-central-1 addressing=virtual-host sse=AES256
  format=jsonl compression=zstd layout=merged align=hour target_mb=128 max_window_ms=300000
  start=latest checkpoint_every=20 memory_mb=1024 fetch_concurrency=4
  discovery_interval_ms=2000 safe_guard_ms=5000 lease_ttl_ms=30000
  multipart_threshold_mb=64 listen=127.0.0.1:9333 crash_at=never

Every enum is printed the way the environment spells it, per-partition and not PerPartition, so the line can be pasted back into a manifest rather than translated. token is <set> or <unset> and never the token; the secret key is never printed at all; and an aws:kms key id is printed as its last four characters, which is enough to confirm the right one was configured and not enough to name the key to a log aggregator.