sink

README

Sink

One gRPC data layer for MongoDB, Elasticsearch, and OpenSearch.

Read, write, and atomically merge documents through a shared API. Sink handles database connections, routing, bounded batching, and backpressure, with optional Kafka-backed asynchronous delivery. Native queries use the same connections.

Quickstart · Documentation · Go client · Releases · Contributing

For independent Store scaling, see Gateway, Engine and Worker and the Compose quickstart.

flowchart LR
  Client --> Gateway
  Gateway --> EngineA[Engine · Store A]
  Gateway --> EngineB[Engine · Store B]
  EngineA --> DatabaseA[(Database A)]
  EngineB --> DatabaseB[(Database B)]
  EngineA --> KafkaA[(Kafka · Store A)]
  KafkaA --> WorkerA[Worker · Store A]
  WorkerA --> DatabaseA

Quickstart

With Docker Compose and Make installed:

git clone https://github.com/liran/sink.git
cd sink
make quickstart

This builds Sink, starts a local MongoDB ReplicaSet and Kafka, and verifies synchronous writes, asynchronous delivery, reads, deletes, and Prometheus metrics. The Go example runs in a container; a local Go installation is not required.

After the checks pass, connect to 127.0.0.1:8080 or inspect metrics. Stop the stack with make quickstart-down. See the quickstart guide for ports, rerunning the example, and resetting local data.

Need just the binary or container? Use the latest release for Linux/macOS binaries and checksums, or run the container.

Why Sink?

Applications that write to several databases often repeat the same non-business work: storage drivers, batching, backpressure, retry rules, read-modify-write conflicts, queue consumers, health checks, and metrics. That logic becomes especially expensive in crawlers and ingestion systems with many concurrent workers.

Sink centralizes those concerns:

Problem	What Sink provides
Each backend has a different API and data model	One batch-native Read, Write, and Delete gRPC API for JSON or BSON documents
Native queries still need direct database clients	Execute, Query, Count, and paged Scan share one command structure and Sink's connections
Every crawler process opens its own database connections	Database connections move into the smaller Sink tier, so connection growth follows Sink replicas instead of crawler processes
Many small calls overload storage	Automatic bounded batching, concurrency limits, and backpressure per store
Some writes must be immediate while others can be buffered	Per-request completion modes, with optional Kafka-backed asynchronous delivery
Concurrent updates lose data	Atomic per-record Lua merges with storage revision checks
A service needs more than one cluster or engine	Named stores selected by the record address, even within one batch
Failures are hard to operate	Per-dependency health, bounded-cardinality metrics, recoverable worker retries, and dead letters

Sink is a good fit when several workers or services need consistent record semantics across shared storage. It is deliberately not an ORM, a schema or index manager, or a cross-record transaction coordinator.

Protect storage behind crawler fleets

A crawler fleet commonly runs many processes, with each process maintaining reusable database connections for its worker threads. If every process writes directly, database connections grow with the crawler fleet while one-record writes add round trips and storage scheduling overhead.

Sink moves that fan-in boundary in front of storage. Workers connect to Sink; each Sink instance owns the backend clients and database connections. Its short, bounded per-store queues coalesce concurrent small RPCs into bulk operations, while concurrency limits and backpressure keep bursts from reaching the database without control.

Database connection demand now follows the deliberately sized Sink tier rather than the crawler process count, and storage receives fewer, fuller requests. This protects the backend from connection and request amplification; it does not replace normal capacity planning. See Architecture and behavior for the exact scope and boundaries.

How it works

Every record has a canonical URI, for example sink://primary/catalog/products/s:product-42. The first component selects a configured Store; only that adapter interprets the path. MongoDB uses database/collection/typed-key; search uses index/typed-key. Gateway hashes the complete URI to select an Engine, consistently across Gateway replicas with the same DNS membership. See record addresses and affinity. Native access uses Execute, Query, Count, and Scan with a common Command.

Every document declares its encoding. MongoDB stores require BSON, so clients apply bson struct tags and retain native BSON values such as datetimes. Elasticsearch and OpenSearch stores require JSON, so clients apply json struct tags and send ordinary JSON without Extended JSON wrappers.

Record requests are batch-native, but a one-record request is the normal single-record form. Sink can combine concurrent small requests into bounded storage batches, preserves the order of operations for the same record, and runs independent records and stores concurrently.

Mutations choose when success is returned:

Completion mode	Success means
WAIT_UNTIL_APPLIED	The backend acknowledged the mutation
WAIT_UNTIL_VISIBLE	A following search read can observe the mutation
RETURN_AFTER_ACCEPTED	The configured Kafka queue durably accepted it

See Architecture and behavior for the full request flow, ordering, batching, merge, and failure semantics.

Connect an application

Go applications can use the typed, concurrency-safe sink-go client:

go get github.com/liran/sink-go

Its quick-start example shows how to connect, create an address, and write a Go value. Other languages can generate a standard gRPC client from proto/sink/sink.proto.

Run the container

Choose the annotated configuration for the component you are deploying:

Component	Configuration	Purpose
Gateway	configs/gateway.yaml	Public ingress, inline Store routes, and forwarding.
Engine	configs/engine.yaml	One Store's synchronous execution and optional asynchronous publication.
Worker	configs/worker.yaml	One Store's Kafka consumption and mutation execution.

For an Engine, copy the role and shared Store examples, edit the Store connection, and mount both files. Clients connect through Gateway.

cp configs/engine.yaml config.yaml
cp configs/stores/primary.yaml store.yaml
# Edit store.yaml for the target backend.
docker run --rm -p 8080:8080 \
  --mount type=bind,source="$(pwd)/config.yaml",target=/etc/sink/config.yaml,readonly \
  --mount type=bind,source="$(pwd)/store.yaml",target=/etc/sink/store.yaml,readonly \
  ghcr.io/batchstream/sink:latest --config /etc/sink/config.yaml --store-config /etc/sink/store.yaml

HTTP health endpoints are always available at health.address (default :8081). Prometheus uses a separate prometheus.address (default :9090) and requires prometheus.enabled: true. Publish the corresponding ports when needed.

Gateway includes its Store routes under forwarding.routes. Restart after changing configuration; routes are loaded only at startup. Worker needs its own configuration and no public gRPC port. The quickstart runs all three roles.

For repeatable deployments, replace latest with a version tag or image digest from the release. See the configuration reference for Store routing and Gateway/Engine/Worker roles, and production sizing for resource budgeting and deployment guidance.

The server validates the complete configuration. Dependency readiness recovers independently; inspect the health of each required store before sending traffic.

Test Lua merge programs

Business applications can validate merge rules locally with the exact Lua runtime used by Sink, without starting a server or storage backend:

sink lua test \
  --script merge/product.lua \
  --cases merge/testdata

The command supports JSON fixtures and type-aware BSON fixtures written as Extended JSON. See Testing Lua merge programs for the case format, direct flags, BSON examples, CI usage, and coverage guidance.

Important guarantees and boundaries

• Atomicity is per record; a multi-record request is not a transaction.
• Operations for one record retain request order. Independent records and stores may execute concurrently.
• Kafka delivery is at least once. A mutation can run again after a worker crashes between applying it and committing its offset. Applications must guarantee business idempotence for every retried or replayed mutation; Sink does not deduplicate business operations. A timeout can leave the outcome unknown.
• Kafka is disabled by default per store. When enabled, Sink owns creation and reconciliation of that store's source and dead-letter Topics at startup.
• Record writes do not initialize indexes, mappings, or aliases. Applications can explicitly initialize them through native Execute commands.
• Dependencies recover independently at startup and runtime. Kafka publication and consumption remain gated until that store's Topic policy is established. Temporary processing failures retain source offsets for automatic recovery.

Documentation

Start with the documentation index, organized by task:

I want to…	Read
Understand routing, batching, and completion	Architecture and document write flow
Query a backend or return an atomic update result	Native access
Write and test a Lua merge	Lua guide and local testing
Deploy and operate Sink	Configuration, sizing, and reliability
Build or contribute	Development and contributing

Contributing and support

Bug reports, documentation improvements, and focused pull requests are welcome. Use the issue forms for bugs and feature proposals, and read CONTRIBUTING.md for local checks and review guidance. Report vulnerabilities through the security policy.

Related projects: Go SDK · Public production qualification suite.

Sink is released under the MIT License.