sink

README

Sink

Sink is a database-independent gRPC service for reading, writing, merging, and deleting explicitly encoded documents. Applications use one record API while Sink handles routing, database connections, batching, concurrency, synchronous or durable asynchronous delivery, and storage-specific behavior for MongoDB, Elasticsearch, and OpenSearch.

Why use 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
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 logical address:

store = primary
namespace = catalog
dataset = products
key = product-42

store selects a configured backend. For MongoDB, namespace and dataset are the database and collection. For Elasticsearch and OpenSearch, dataset is the complete existing index or alias name. The application never sends a database connection string or storage-specific query through the API.

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.

All 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.

Try it locally

The quickstart requires Docker with Compose. From the repository root, run:

make quickstart

This builds Sink, starts MongoDB and Kafka, runs synchronous and asynchronous record operations, checks Prometheus metrics, and leaves the stack available at 127.0.0.1:8080.

make quickstart-down

The quickstart guide lists the exposed ports, direct Compose commands, and reset instructions.

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.

For a real deployment, copy config.example.yaml, edit the backend connection, and start the container with the file mounted:

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

The server validates the complete configuration and connects to every required dependency before becoming ready.

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.
• Sink does not create Elasticsearch or OpenSearch indexes, mappings, or aliases.
• 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

• Reliability and recovery — idempotence responsibility, fault handling, capacity, durability baselines, alerting, and DLQ replay

• Architecture and behavior — request flow, adapters, batching, Lua merges, asynchronous delivery, and reliability boundaries

• Lua merge developer guide — script contract, built-in sink.v1 tools, examples, retry semantics, and testing guidance

• Testing Lua merge programs — local production-parity runner, JSON/BSON fixtures, case suites, and CI integration

• Configuration reference — every field, default, allowed value, validation rule, routing rule, and deployment mode

• Docker Compose quickstart — local environment and end-to-end example

• Development guide — repository layout, validation, and release images

• Protocol definition — authoritative gRPC contract

Sink is released under the MIT License.