A change-data-capture service that reads from a PostgreSQL replication slot and writes to a search index has two unforgiving constraints. It cannot stop reading for long, because PostgreSQL retains write-ahead log segments for lagging slots. It also cannot absorb pressure by buffering indefinitely, because a Go heap full of transient events, JSON buffers, maps, and slices turns into garbage-collector work and tail-latency spikes.
The reusable engineering pattern is a bounded streaming pipeline: read changes continuously, transform them with allocation-aware code, batch them for the downstream search system, and apply explicit backpressure when the indexer cannot keep up. The goal is not to maximize one benchmark number. The goal is stable memory, stable lag, and predictable recovery behavior under sustained volume.
This topic focuses on the Go-specific implementation decisions behind that pattern: bounded queues, limited in-flight bulk requests, reusable buffers with sync.Pool, JSON encoder validation, and garbage-collector tuning only after profiling. The same design applies to CDC fan-out, audit-log indexing, event enrichment, search materialization, and other services that bridge a relentless source to a slower sink.
A replication slot keeps producing changes as long as the primary database accepts writes. If the consumer stalls, the database retains more WAL. If the consumer reads too eagerly, memory moves from the database into the service. A bounded pipeline makes that tradeoff explicit by limiting queue length, batch size, and concurrent downstream writes.
High-volume Go services often fail through death by small allocations. Per-event structs, temporary maps, JSON byte slices, and batch buffers are short lived, but the garbage collector still has to discover and reclaim them. Reuse is useful only when the object is easy to reset and does not carry hidden ownership, locks, contexts, or references to old request data.
Replacing encoding/json with a faster encoder such as jsoniter can reduce CPU and allocations for repetitive payloads, but it changes the risk profile. Compatibility modes, null handling, custom marshalers, omitted fields, and third-party nullable types need regression tests because search documents are externally visible data contracts.
Backpressure is not just a performance knob. It prevents a CDC bridge from converting downstream slowness into unbounded heap growth. A full queue should slow reads, reduce batch admission, or surface lag metrics rather than silently allocating more memory.
The ingest loop should make capacity visible. A common shape is one goroutine reading changes, a fixed number of workers transforming events, and a bulk indexer with a semaphore limiting concurrent requests. The bounded channel is the handoff contract: once it is full, upstream pressure is real.
type Change struct {
Table string
Key string
Data []byte
}
type BulkOp struct {
Action string
ID string
Body []byte
}
func RunPipeline(ctx context.Context, src ChangeReader, sink BulkIndexer) error {
changes := make(chan Change, 4096)
ops := make(chan BulkOp, 4096)
g, ctx := errgroup.WithContext(ctx)
g.Go(func() error {
defer close(changes)
for {
ch, err := src.Next(ctx)
if err != nil {
return err
}
select {
case changes <- ch:
case <-ctx.Done():
return ctx.Err()
}
}
})
for i := 0; i < 8; i++ {
g.Go(func() error {
for ch := range changes {
op, err := transform(ch)
if err != nil {
return err
}
select {
case ops <- op:
case <-ctx.Done():
return ctx.Err()
}
}
return nil
})
}
g.Go(func() error {
defer close(ops)
<-ctx.Done()
return nil
})
g.Go(func() error {
return writeBatches(ctx, sink, ops, 5_000, 5*time.Millisecond)
})
return g.Wait()
}sync.Pool is appropriate for temporary buffers shared by many independent goroutines. The pool is allowed to drop objects at any time, so correctness cannot depend on reuse. Treat pooled objects as an optimization boundary: get, reset, use, copy out anything that must survive, reset again, and put back.
var bulkBufferPool = sync.Pool{
New: func() any {
b := bytes.NewBuffer(make([]byte, 0, 256<<10))
return b
},
}
func encodeBulk(ops []BulkOp) []byte {
buf := bulkBufferPool.Get().(*bytes.Buffer)
buf.Reset()
defer func() {
if buf.Cap() <= 2<<20 {
bulkBufferPool.Put(buf)
}
}()
for _, op := range ops {
writeActionLine(buf, op.Action, op.ID)
buf.Write(op.Body)
buf.WriteByte('\n')
}
out := make([]byte, buf.Len())
copy(out, buf.Bytes())
return out
}The most direct use case is streaming database mutations into Elasticsearch, OpenSearch, Meilisearch, or another query-optimized store. The database remains the source of truth while the search cluster receives bounded, batched updates that are tuned for indexing throughput.
The same structure works when each change event must be enriched with reference data, normalized, scored, or filtered before publication. The important rule is that enrichment does not get to create an unbounded side queue when a downstream dependency slows down.
CDC pipelines need service-level objectives for replication lag, in-flight operations, heap size, and bulk-index latency. Those metrics tell operators whether to scale workers, reduce batch size, shed optional enrichment, or pause writes safely before WAL retention becomes a database incident.
Runtime flags and experimental garbage collectors should be late-stage controls. First remove accidental allocations, stabilize queue sizes, and benchmark the real serialization path. Only then compare GC behavior with representative traffic, because a lower GC CPU percentage is not useful if lag, memory, or downstream retry rates get worse.
func BenchmarkTransform(b *testing.B) {
sample := loadRepresentativeChange()
b.ReportAllocs()
for i := 0; i < b.N; i++ {
op, err := transform(sample)
if err != nil {
b.Fatal(err)
}
_ = op
}
}
// Compare under the same payload corpus:
// go test -bench=Transform -benchmem ./...
// GOEXPERIMENT=greenteagc go test -bench=Transform -benchmem ./...A JSON encoder switch should ship with golden tests that compare required fields, null behavior, omitted fields, and representative search documents. Benchmark wins are only acceptable when the indexed document contract is unchanged or intentionally migrated.