Bridge AWS Kinesis to Tessera

Planned (Phase 3)

Planned for Phase 3. This guide describes the planned setup; the configuration shown below is validated against the current API schema so it stays accurate.

tessera/bus-bridge is the same lightweight container planned for Kafka, run next to your Kinesis stream: it reads every shard with KCL-style leases kept in a DynamoDB table you own, maps each JSON record to a canonical event with the same declarative mapping config, and delivers batches to POST /v1/events — at-least-once from Kinesis, exactly-once into Tessera via idempotent event ids derived from (bridge_id, stream, shard id, sequence number). Your AWS account, stream and credentials stay yours; Tessera has no AWS footprint.

Prerequisites

An API key with the ingest scope.
A Kinesis stream carrying JSON records, and somewhere to run a container (EKS, ECS, EC2, …) with access to Kinesis, DynamoDB and ingest.staging.tessera.zone.
Producer aggregation disabled. KPL-aggregated records are protobuf blobs, not JSON: they cannot be mapped and land in rejects. Turn off aggregation in the KPL (AggregationEnabled=false) or produce with the AWS SDK's PutRecord(s).
Use the entity id as the partition key. Each shard is delivered in order through its own queue, so per-entity order is preserved exactly when one entity always maps to one shard — the Kinesis analog of partitioning a Kafka topic by entity id.
One bridge (one bridge_id, one checkpoint table) per stream. Run more replicas of it to scale; never point two different bridges at the same table.

Create the credentials (least privilege)

The bridge resolves credentials with the default AWS provider chain — AWS_ACCESS_KEY_ID/AWS_SECRET_ACCESS_KEY environment variables, EKS IAM Roles for Service Accounts (IRSA: AWS_ROLE_ARN + AWS_WEB_IDENTITY_TOKEN_FILE, injected for you by EKS), or an instance profile. Credentials are never read from the config file (it rejects credential-looking keys). On EKS, prefer IRSA: annotate the bridge's service account with the role below.

bridge-iam-policy.json
{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "ReadSourceStream",
      "Effect": "Allow",
      "Action": [
        "kinesis:DescribeStreamSummary",
        "kinesis:ListShards",
        "kinesis:GetShardIterator",
        "kinesis:GetRecords"
      ],
      "Resource": "arn:aws:kinesis:us-east-1:111122223333:stream/viewer-activity"
    },
    {
      "Sid": "PublishRejects",
      "Effect": "Allow",
      "Action": ["kinesis:PutRecord"],
      "Resource": "arn:aws:kinesis:us-east-1:111122223333:stream/viewer-activity-rejects"
    },
    {
      "Sid": "LeaseAndCheckpointTable",
      "Effect": "Allow",
      "Action": [
        "dynamodb:DescribeTable",
        "dynamodb:GetItem",
        "dynamodb:PutItem",
        "dynamodb:UpdateItem"
      ],
      "Resource": "arn:aws:dynamodb:us-east-1:111122223333:table/viewer-activity-bridge-checkpoints"
    }
  ]
}

Drop the PublishRejects statement when you do not configure a rejects stream. Add dynamodb:CreateTable to the table statement only if you set create_checkpoint_table: true.

Provision the checkpoint table

The table holds one item per shard: the lease (which replica reads it) and the checkpoint (the last sequence number Tessera accepted). Create it once — on demand capacity is plenty — or let the bridge create it with create_checkpoint_table: true:

checkpoint-table.cfn.yaml
Resources:
  BridgeCheckpoints:
    Type: AWS::DynamoDB::Table
    Properties:
      TableName: viewer-activity-bridge-checkpoints
      BillingMode: PAY_PER_REQUEST
      AttributeDefinitions:
        - { AttributeName: shard_id, AttributeType: S }
      KeySchema:
        - { AttributeName: shard_id, KeyType: HASH }

A missing table fails startup with an error naming this schema. Items are never deleted or TTL-expired — checkpoint history must outlive leases.

Attribute	Type	Meaning
`shard_id`	S (key)	Kinesis shard id
`lease_owner`	S	replica holding the lease (absent while free)
`lease_counter`	N	increments on every acquisition — fences stale replicas
`lease_expiry`	N	epoch seconds; renewed every TTL/3
`checkpoint`	S	last accepted sequence number, `SHARD_END`, or absent
`parent_shard_ids`	L	parents after a split/merge
`updated_at`	N	epoch seconds of the last write
`stream`, `bridge_id`	S	scope guard: a table serves one bridge and stream
`latest_start_epoch`	N	`start_position: latest` boundary (optional)

Write the config

The mapping block is identical to the planned Kafka bridge's — any mapping that bridge accepts, the Kinesis bridge accepts. Worked example: viewer entities emitting playback.started / playback.stopped records on viewer-activity.

bus-bridge.yaml
bridge_id: viewer-activity-bridge # keep stable: part of derived event ids
consumer:
  kind: kinesis
  region: us-east-1
  stream: viewer-activity
  checkpoint_table: viewer-activity-bridge-checkpoints
  start_position: trim_horizon # trim_horizon | latest | at_timestamp (+ start_timestamp)
  rejects_stream: viewer-activity-rejects # optional; null → count + log only
  max_in_flight_events: 5000 # one budget shared by all shards
  lease_ttl_seconds: 30 # failover within 3x TTL
mapping:
  version: 1
  encoding: json
  entity_type: viewer
  entity_id_pointer: /viewer/id
  event_type_source: { pointer: /type }
  timestamp_pointer: /occurred_at # RFC 3339 or epoch seconds/millis
  event_id_pointer: null # derive ids from stream + shard + sequence number
  attributes: passthrough_flat

Validate it before deploying — no credentials and no AWS calls are needed:

docker run --rm -v $PWD/bus-bridge.yaml:/etc/tessera/bus-bridge.yaml:ro \
  tessera/bus-bridge:0.2.0 --check-config

start_position applies to shards with no checkpoint yet. latest is pinned the first time a shard starts, so a restart before the first checkpoint still resumes from that point — nothing written meanwhile is skipped.

Run the bridge

docker run -d --name tessera-bus-bridge -e TESSERA_API_KEY -p 8080:8080 \
  -v $PWD/bus-bridge.yaml:/etc/tessera/bus-bridge.yaml:ro \
  tessera/bus-bridge:0.2.0 # pin a semver tag — never latest

services/bus-bridge/docker-compose.kinesis.example.yaml in the repository is a complete example. Run 2+ replicas with the same config for availability: replicas divide the shards through the lease table, a replica holding more than its share yields shards to a newcomer (steal-based rebalancing), a crashed replica's shards are taken over within 3× lease_ttl_seconds, and SIGTERM drains in-flight batches, checkpoints and releases its leases for immediate handover.

Delivery semantics

Checkpoints — a shard's checkpoint only advances past records Tessera accepted (or that were terminally rejected), at most every 5 s or 1,000 records, at shard end and on shutdown. A crash replays from the last checkpoint; replayed events dedup on their derived event_id.
Resharding — after a split or merge, children start only once every parent is checkpointed SHARD_END, so per-partition-key order survives resharding. Discovery runs every reshard_poll_seconds.
Backpressure — rate limits (429) and outages stop GetRecords on every shard (bounded in-flight budget, default 5,000 events) while leases keep renewing; the bridge rides out long ingest outages without loss or lease churn. Reads respect Kinesis limits (≤ 5 calls/s and 2 MiB/s per shard).
Rejects — unmappable records (missing entity id, an entity id that is not an identifier such as true, 12.5, NaN or a whole number above 2^53 − 1 sent as a JSON number, an entity id with #, bad timestamp, an event_id with # in its first 16 characters, non-JSON) and per-item ingest rejections are counted, logged with {stream, shard_id, sequence_tail, sequence_sha, partition_key} (a log line never carries the whole sequence number: sequence_tail is its last 12 digits and sequence_sha the first 12 hex characters of its SHA-256), and — with rejects_stream — published there with the original partition key as a JSON envelope (failure, coordinates, bridge_id, occurred_at, payload_base64). A record too large to fit base64-encoded in a Kinesis record is published without payload_base64 (its size and SHA-256 instead). Rejects never block the stream.
Secondary ids — a participant id that is not an identifier (the rule), or is longer than 256 characters, never rejects the record: the event is delivered without that participant and the drop is counted in bridge_secondary_dropped_total{topic,reason}.
Retention — a checkpoint older than the stream's retention resumes at the oldest available record; the gap is counted in bridge_data_expired_total and logged.

Monitor it

/healthz (liveness: shard workers alive and leases renewing) and /metrics (Prometheus): bridge_delivered_total, bridge_rejected_total{reason}, bridge_secondary_dropped_total{topic,reason}, bridge_retry_total, bridge_rekey_total, bridge_paused, bridge_delivery_latency_seconds, plus bridge_shard_lag_ms{shard_id} (from MillisBehindLatest), bridge_leases_held, bridge_checkpoint_age_seconds{shard_id} and bridge_data_expired_total. Import services/bus-bridge/dashboards/kinesis.json into Grafana for lag, leases, checkpoint age and delivery/reject rates.

Troubleshooting

Symptom	Cause	Fix
startup fails naming the checkpoint-table schema	the table is missing or has another key	create it as above, or set `create_checkpoint_table: true`
`holds shard … for bridge …` at startup	the table belongs to another bridge or stream	use one table per `bridge_id` and stream
`bridge_paused` stuck at 1	ingest 429s or an outage	check `GET /usage/quotas`; the bridge catches up on its own
everything lands in rejects as `not_json`	KPL aggregation is on	disable producer aggregation
`bridge_shard_lag_ms` keeps growing	too few replicas or a slow ingest	add replicas (up to one per shard) and check `bridge_retry_total`
`bridge_data_expired_total` increased	a checkpoint or parent aged out of retention	raise the stream's retention or keep the bridge running

Prerequisites​

Create the credentials (least privilege)​

Provision the checkpoint table​

Write the config​

Run the bridge​

Delivery semantics​

Monitor it​

Troubleshooting​