cdcfresh

README

cdcfresh

CDC-powered freshness for derived tables in Go.

Keep rollup/lookup/read-model tables fresh without SQL triggers, stored procedures, or cron-scan waste: change-data-capture events are the doorbell, your own SQL is the recipe, cdcfresh is the orchestration in between.

Status: v1 is feature-complete — core loop and the TiCDC-on-Pulsar source adapter both land and are covered end to end against real infrastructure. The API is not yet tagged, so it may still shift. Full model and guarantees are in the package documentation.

src, err := pulsar.Source(
	"pulsar://localhost:6650",
	[]string{"persistent://public/default/cdcfresh"},
	pulsar.WithSubscription("device-totals"),
)
if err != nil {
	return err
}
defer src.Close()

r, err := cdcfresh.New(
	cdcfresh.Source(src),

	// Which derived-table scopes did this change dirty?
	cdcfresh.Scope(func(ev cdcfresh.RowEvent) []cdcfresh.Key {
		device, _ := ev.Data["device"].(string)
		return []cdcfresh.Key{cdcfresh.Key(device)}
	}),

	// Recompute one scope from the source tables. Your SQL, and nothing
	// from the event reaches it — the event only named the scope.
	cdcfresh.Rebuild(func(ctx context.Context, k cdcfresh.Key) error {
		_, err := db.ExecContext(ctx,
			`REPLACE INTO device_totals (device, total)
			 SELECT device, SUM(value) FROM readings WHERE device = ? GROUP BY device`,
			string(k))
		return err
	}),

	cdcfresh.Coalesce(5*time.Second),
	cdcfresh.MaxWait(30*time.Second),
	cdcfresh.Reconcile(time.Hour, allDevices),
)
if err != nil {
	return err
}
return r.Run(ctx)

The pattern

Databases like TiDB have no triggers, procedures, or event schedulers — and even where triggers exist, in-transaction delta maintenance drifts and taxes every write. The robust alternative is a loop that every team keeps re-implementing by hand:

CDC stream ──► extract dirty scope keys ──► coalesce ──► scoped rebuild-from-truth
   (TiCDC)         (your function)          (debounce)      (your SQL, idempotent)
                                                     └──► reconcile sweep (safety net)

The events are only a doorbell, never the data: rebuilds always recompute the affected scope from the source tables, so duplicates are harmless (at-least-once friendly) and drift is impossible by construction.

Design principles

• Ephemeral / stateless — cdcfresh owns no durable state. The dirty-scope queue is in memory; stream position is the broker subscription cursor; a crash loses nothing that redelivery + the reconcile sweep don't heal. Embeddable in any Go service the way testcontainers is embeddable in any test suite: import, configure in a few lines, run.
• Your SQL stays yours — the library never generates or manages queries, tables, or schemas. It decides when and for which scope to run what you wrote.
• At-least-once native — every callback must be (and is documented as) idempotent; duplicate events are a non-event.
• Start small — v1 is a prototype for exactly one source technology: TiCDC → Pulsar (canal-json). The source is behind a tiny interface so MySQL binlog / Kafka adapters can come later, but they are explicitly out of scope for v1.

v1 scope

In	Out (v1)
Pulsar consumer for TiCDC canal-json events	Kafka / MySQL-binlog / Postgres sources
Scope extraction, per-key coalescing/debounce	Exactly-once guarantees
Rebuild invocation with retry + backoff	Generating rebuild SQL
Reconcile sweep scheduler	Managing schemas or migrations
Lag/health counters via a Stats() snapshot	Serving reads, HTTP anything
Integration tests against real Pulsar and TiDB containers	Multi-instance coordination (single consumer assumed; document failover subscription)

Repository layout

cdcfresh/            root package — import "github.com/bis-code/cdcfresh"
├── event.go         public contract: Key, RowEvent, Event, EventSource, ErrSkip
├── options.go       Option constructors + validation
├── refresher.go     Refresher, New, Run
├── loops.go         receive / schedule / worker / reconcile goroutines
├── coalesce.go      dirty-set state machine (pure, explicit clock)
├── backoff.go       retry delay
├── stats.go         atomic counters + Stats snapshot
├── internal/
│   ├── canaljson/   canal-json decoder + fixtures captured from a real TiCDC
│   └── testenv/     integration-tier containers: one Pulsar, one TiDB
├── pulsar/          Pulsar EventSource adapter (the only package with a client)
└── test/cdcstack/   full TiDB + TiCDC + Pulsar stack for local development

The root package links nothing outside the standard library, and CI proves it with go list -deps .. Adapters and tests take dependencies freely; none of them may become reachable from the root.

Testing

make test              # every tier
make test-unit         # unit tier only — no Docker
make test-integration  # integration tier — needs Docker

Integration tests start the containers they need and stop them again, so there is nothing to bring up first.

Observability

Stats() returns a plain snapshot struct — no metrics client in the library, so it costs nothing and dictates nothing. Wire it to whatever you already run:

// expvar, standard library
expvar.Publish("cdcfresh", expvar.Func(func() any { return r.Stats() }))

// Prometheus: set gauges from the same fields on scrape
prometheus.NewGaugeFunc(prometheus.GaugeOpts{Name: "cdcfresh_dirty_keys"},
	func() float64 { return float64(r.Stats().DirtyKeys) })

DirtyKeys and PoisonedKeys are queue depth; EventsReceived, EventsSkipped, RebuildsOK, RebuildsFailed and Reconciles are counters; LastEvent and LastRebuild are timestamps for staleness alerting.

Prior art / positioning

• Debezium — the CDC heavyweight, Java, owns capture + delivery; cdcfresh starts after delivery and stays a library.
• Materialize / Readyset — "we run your views" systems; cdcfresh packages a pattern instead of hosting your queries.
• go-mysql, watermill, tiflow internals — parts of the loop exist; the doorbell→dirty-scope→rebuild orchestration as an embeddable package does not.

License

MIT — see LICENSE.

Documentation

Overview

Package cdcfresh keeps derived tables fresh from change-data-capture events: CDC events are the doorbell, your SQL is the recipe, cdcfresh is the orchestration in between.

Model

A Refresher runs one loop: consume decoded CDC events from a Source, extract the dirty scope keys each event implies (your Scope function), coalesce repeated dirties per key with a debounce-plus-max-wait cap, and hand each ready key to your Rebuild callback, which recomputes that scope from the source tables. Events never carry decoded column data into your SQL — they only say which key is dirty. That is what makes duplicate events harmless (Rebuild just runs again) and drift impossible: the derived table is always recomputed from truth, never patched in place.

Usage

Construct a Refresher with New and run it until ctx is cancelled:

refresher, err := cdcfresh.New(
	cdcfresh.Source(src), // an EventSource, e.g. from cdcfresh/pulsar
	cdcfresh.Scope(func(ev cdcfresh.RowEvent) []cdcfresh.Key {
		return []cdcfresh.Key{cdcfresh.Key(ev.Database + "." + ev.Table)}
	}),
	cdcfresh.Rebuild(func(ctx context.Context, k cdcfresh.Key) error {
		return rebuildScope(ctx, k) // your idempotent SQL, scoped to k
	}),
	cdcfresh.Coalesce(5*time.Second),
	cdcfresh.MaxWait(30*time.Second),
	cdcfresh.Workers(4),
	cdcfresh.OnError(func(err error) { log.Println(err) }),
)
if err != nil {
	log.Fatal(err)
}
err = refresher.Run(ctx) // blocks until ctx is done or the source fails

Source, Scope, and Rebuild are required; New reports every missing or invalid option in a single error rather than failing on the first one. Reconcile, Backoff, and PoisonAfter round out the tuning surface — see Failures and Reconcile below.

Guarantees

cdcfresh is at-least-once, not exactly-once: Scope and Rebuild must both be idempotent, because retries, redelivery, and the reconcile sweep all invoke them more than once for the same event or key. Coalescing is a per-key debounce with a max-wait cap — a key fires once it has been quiet for Coalesce or dirty for MaxWait, whichever comes first — never a fixed window. At most one rebuild is ever in flight per key; an event that arrives while a key's rebuild is running re-dirties it, and the key re-enters the queue once that rebuild finishes, so no event is ever lost to a race with an in-progress rebuild.

A source event is acknowledged as soon as its keys are enqueued, not after the rebuild that eventually processes them completes. A crash between those two points loses only the in-memory dirty set — the reconcile sweep (below) heals it. This is what keeps the library stateless: the broker's subscription cursor is the only durable position, and cdcfresh never needs one of its own.

Failures

A failed rebuild retries with exponential backoff and jitter (see Backoff), and one failing key never blocks any other key's progress. After PoisonAfter consecutive failures a key is quarantined: removed from the retry loop, reported once through OnError, and counted in Stats.PoisonedKeys. Only a reconcile sweep re-admits a poisoned key, and a success there resets its failure count.

Reconcile

Reconcile(every, enumerate) adds a periodic healing sweep: enumerate lists the live key universe, and every key it returns flows through the normal pipeline — the same debounce, single-flight, and poison-healing logic as CDC-origin keys, with no separate code path. One sweep runs immediately when Run starts, then again every interval. The startup sweep is what heals a failover takeover: a standby that resumes from the broker's last acked position picks up any key that was enqueued but not yet rebuilt on the instance it replaced.

Deployment

cdcfresh assumes a single active consumer per Refresher — for a Pulsar source, a failover subscription, with standbys running the same binary idle. All state (the dirty set, backoff timers, the poison list) lives in memory and dies with the process; a takeover is healed by broker redelivery plus the startup reconcile sweep, not by any coordination cdcfresh performs itself. A Refresher is single-use: Run returns an error immediately if called a second time.

Observability

Stats returns a point-in-time snapshot of the loop's counters (events received/skipped, keys marked, rebuilds ok/failed, reconcile sweeps) and gauges (dirty keys, poisoned keys). Counters and gauges are sampled independently, so the snapshot is not a single atomic cut across every field. cdcfresh has no metrics dependency of its own — publishing a Stats snapshot through expvar or a Prometheus collector is a few lines on the caller's side.

Example

Example wires a Refresher to a source and watches three row changes across two devices collapse into two rebuilds — one per dirty scope, not one per event.

package main

import (
	"context"
	"fmt"
	"log"
	"sort"
	"time"

	"github.com/bis-code/cdcfresh"
)

// memSource is a stand-in for a real adapter, so this example stays runnable
// without a broker. It blocks once drained, as a real source does between
// events — returning an error there would stop Run.
type memSource struct{ events []cdcfresh.Event }

func (s *memSource) Receive(ctx context.Context) (cdcfresh.Event, error) {
	if len(s.events) == 0 {
		<-ctx.Done()
		return cdcfresh.Event{}, ctx.Err()
	}
	ev := s.events[0]
	s.events = s.events[1:]
	return ev, nil
}

func change(device string) cdcfresh.Event {
	return cdcfresh.Event{Row: cdcfresh.RowEvent{
		Database: "shop",
		Table:    "readings",
		Type:     cdcfresh.Insert,
		Data:     map[string]any{"device": device},
	}}
}

func main() {
	src := &memSource{events: []cdcfresh.Event{
		change("dev-a"), change("dev-a"), change("dev-b"),
	}}

	rebuilt := make(chan cdcfresh.Key, 4)

	r, err := cdcfresh.New(
		cdcfresh.Source(src),

		// Scope names the derived-table scopes a change dirties. It runs
		// inline in the receive loop, so keep it pure and fast.
		cdcfresh.Scope(func(ev cdcfresh.RowEvent) []cdcfresh.Key {
			device, _ := ev.Data["device"].(string)
			return []cdcfresh.Key{cdcfresh.Key("device:" + device)}
		}),

		// Rebuild recomputes one scope from the source tables. It must be
		// idempotent: at-least-once delivery makes repeats routine, and the
		// event is only a doorbell — none of its values belong in this SQL.
		cdcfresh.Rebuild(func(ctx context.Context, k cdcfresh.Key) error {
			rebuilt <- k
			return nil
		}),

		cdcfresh.Coalesce(200*time.Millisecond),
		cdcfresh.MaxWait(2*time.Second),
	)
	if err != nil {
		log.Fatal(err)
	}

	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	go r.Run(ctx)

	// Two scopes were dirtied by three events; the two dev-a changes coalesce.
	keys := []string{string(<-rebuilt), string(<-rebuilt)}
	sort.Strings(keys)
	fmt.Println(keys)

}

Output:
[device:dev-a device:dev-b]

Index

• Variables
• type EnumerateFunc
• type Event
• type EventSource
• type EventType
  • func (t EventType) String() string
• type Key
• type Option
  • func Backoff(base, maxDelay time.Duration) Option
  • func Coalesce(d time.Duration) Option
  • func MaxWait(d time.Duration) Option
  • func OnError(f func(error)) Option
  • func PoisonAfter(n int) Option
  • func Rebuild(f RebuildFunc) Option
  • func Reconcile(every time.Duration, enumerate EnumerateFunc) Option
  • func Scope(f ScopeFunc) Option
  • func Source(s EventSource) Option
  • func Workers(n int) Option
• type RebuildFunc
• type Refresher
  • func New(opts ...Option) (*Refresher, error)
  • func (r *Refresher) Run(ctx context.Context) error
  • func (r *Refresher) Stats() Stats
• type RowEvent
• type ScopeFunc
• type Stats

Examples

• Package
• Reconcile
• Refresher.Stats

Variables

var ErrSkip = errors.New("cdcfresh: event skipped")

ErrSkip marks a Receive error as a non-fatal, per-event skip rather than a fatal source failure. An adapter wraps a per-event error it cannot recover from (e.g. an undecodable message) with it — fmt.Errorf("%w: ...", cdcfresh.ErrSkip) — and acks the dropped message itself, since it holds the ack handle. The core counts the skip, reports it via OnError, and continues consuming.

Types

type EnumerateFunc

type EnumerateFunc func(context.Context) ([]Key, error)

EnumerateFunc lists the live key universe for a reconcile sweep.

type Event

type Event struct {
	Row RowEvent
	Ack func()
}

Event pairs a RowEvent with its acknowledgement handle. Ack (if non-nil) is called by the core once the event's keys are enqueued — before, and independent of, any rebuild.

type EventSource

type EventSource interface {
	Receive(ctx context.Context) (Event, error)
}

EventSource delivers decoded events. Receive blocks until an event is available or ctx is done. Errors are fatal to Run unless they wrap ErrSkip: adapters own transient-failure retry internally.

type EventType

type EventType uint8

EventType classifies a row change.

const (
	// Unknown is the zero value: an adapter that fails to set Type produces
	// this rather than a misleading concrete kind.
	Unknown EventType = 0
	Insert  EventType = iota
	Update
	Delete
)

func (EventType) String

func (t EventType) String() string

String renders the event type for logs and error messages.

type Key

type Key string

Key identifies one dirty scope of a derived table. It is opaque to the library: encode any structure you need ("device:123") and parse it only inside your own Rebuild.

type Option

type Option func(*config)

Option configures a Refresher. Source, Scope, and Rebuild are required; New reports every missing one in a single error.

func Backoff

func Backoff(base, maxDelay time.Duration) Option

Backoff sets the retry schedule after rebuild failures: base doubles per consecutive failure, capped at maxDelay, jittered.

func Coalesce

func Coalesce(d time.Duration) Option

Coalesce sets the per-key quiet period before a dirty key fires.

func MaxWait

func MaxWait(d time.Duration) Option

MaxWait caps how long a continuously-dirty key may wait before firing.

func OnError

func OnError(f func(error)) Option

OnError receives skipped-event, rebuild, poison, and reconcile errors.

func PoisonAfter

func PoisonAfter(n int) Option

PoisonAfter quarantines a key after n consecutive rebuild failures; only a reconcile sweep re-admits it.

func Rebuild

func Rebuild(f RebuildFunc) Option

Rebuild sets the per-key rebuild callback (required).

func Reconcile

func Reconcile(every time.Duration, enumerate EnumerateFunc) Option

Reconcile schedules a healing sweep: enumerate lists the live key universe; every key flows through the normal pipeline. One sweep runs at Run start, then every interval.

Example

ExampleReconcile schedules the healing sweep. Enumerate lists the live key universe; every key it returns flows through the normal coalescing pipeline, which is also how a quarantined key is re-admitted.

package main

import (
	"context"
	"log"
	"time"

	"github.com/bis-code/cdcfresh"
)

// memSource is a stand-in for a real adapter, so this example stays runnable
// without a broker. It blocks once drained, as a real source does between
// events — returning an error there would stop Run.
type memSource struct{ events []cdcfresh.Event }

func (s *memSource) Receive(ctx context.Context) (cdcfresh.Event, error) {
	if len(s.events) == 0 {
		<-ctx.Done()
		return cdcfresh.Event{}, ctx.Err()
	}
	ev := s.events[0]
	s.events = s.events[1:]
	return ev, nil
}

func main() {
	devices := func(ctx context.Context) ([]cdcfresh.Key, error) {
		// SELECT DISTINCT the scopes that should exist, in practice.
		return []cdcfresh.Key{"device:dev-a", "device:dev-b"}, nil
	}

	_, err := cdcfresh.New(
		cdcfresh.Source(&memSource{}),
		cdcfresh.Scope(func(cdcfresh.RowEvent) []cdcfresh.Key { return nil }),
		cdcfresh.Rebuild(func(context.Context, cdcfresh.Key) error { return nil }),

		// One sweep runs at Run start — which is what heals a failover
		// takeover — and one every interval after it.
		cdcfresh.Reconcile(time.Hour, devices),
	)
	if err != nil {
		log.Fatal(err)
	}
}

func Scope

func Scope(f ScopeFunc) Option

Scope sets the event→keys mapping (required).

func Source

func Source(s EventSource) Option

Source sets the event source (required).

func Workers

func Workers(n int) Option

Workers bounds concurrent rebuilds.

type RebuildFunc

type RebuildFunc func(context.Context, Key) error

RebuildFunc recomputes one scope from the source tables. It must be idempotent: at-least-once delivery makes duplicate invocations routine.

type Refresher

type Refresher struct {
	// contains filtered or unexported fields
}

Refresher orchestrates the CDC→coalesce→rebuild loop. Create with New, start with Run.

func New

func New(opts ...Option) (*Refresher, error)

New validates options and builds a Refresher. It reports every missing required option, and every option given an invalid value, in one error.

func (*Refresher) Run

func (r *Refresher) Run(ctx context.Context) error

Run consumes the source, coalesces dirty keys, and drives rebuilds until ctx is cancelled (returns ctx.Err()) or the source fails (returns the wrapped error). A Refresher is single-use: call Run once; a second call returns an error without starting anything.

func (*Refresher) Stats

func (r *Refresher) Stats() Stats

Stats returns a snapshot of the loop's counters and gauges; safe to call from any goroutine.

Example

ExampleRefresher_Stats publishes the counters without pulling a metrics client into the library. Stats is a plain snapshot struct, so any collector can read it: this uses expvar from the standard library, and a Prometheus collector would set gauges from the same fields on scrape.

package main

import (
	"context"
	"expvar"
	"fmt"
	"log"

	"github.com/bis-code/cdcfresh"
)

// memSource is a stand-in for a real adapter, so this example stays runnable
// without a broker. It blocks once drained, as a real source does between
// events — returning an error there would stop Run.
type memSource struct{ events []cdcfresh.Event }

func (s *memSource) Receive(ctx context.Context) (cdcfresh.Event, error) {
	if len(s.events) == 0 {
		<-ctx.Done()
		return cdcfresh.Event{}, ctx.Err()
	}
	ev := s.events[0]
	s.events = s.events[1:]
	return ev, nil
}

func main() {
	r, err := cdcfresh.New(
		cdcfresh.Source(&memSource{}),
		cdcfresh.Scope(func(cdcfresh.RowEvent) []cdcfresh.Key { return nil }),
		cdcfresh.Rebuild(func(context.Context, cdcfresh.Key) error { return nil }),
	)
	if err != nil {
		log.Fatal(err)
	}

	expvar.Publish("cdcfresh", expvar.Func(func() any { return r.Stats() }))

	s := r.Stats()
	fmt.Println(s.EventsReceived, s.RebuildsOK, s.DirtyKeys)

}

Output:
0 0 0

type RowEvent

type RowEvent struct {
	Database string
	Table    string
	Type     EventType
	PKNames  []string
	Data     map[string]any // new row image (Insert/Update)
	Old      map[string]any // previous row image (Update/Delete)

	// CommitTs is a source-defined ordering token, monotonic within one
	// source (e.g. a TiCDC TSO, a Debezium ts_ms, a binlog position). It is
	// not wall-clock time and not comparable across sources.
	CommitTs uint64
}

RowEvent is a decoded change event — a doorbell, never data: cdcfresh forwards it to Scope and nothing else.

type ScopeFunc

type ScopeFunc func(RowEvent) []Key

ScopeFunc maps a row event to the derived-table scopes it dirties. It runs inline in the receive loop: keep it pure and fast — a slow Scope throttles receive, and therefore ack (D6).

type Stats

type Stats struct {
	EventsReceived uint64
	EventsSkipped  uint64
	KeysMarked     uint64
	RebuildsOK     uint64
	RebuildsFailed uint64
	Reconciles     uint64
	DirtyKeys      int
	PoisonedKeys   int
	LastEvent      time.Time // zero if no event has been received yet
	LastRebuild    time.Time // zero if no rebuild has completed yet
}

Stats is a point-in-time snapshot of the loop's counters and gauges. Counters and gauges are sampled independently — counters are atomic reads, gauges (DirtyKeys, PoisonedKeys) are read under the dirty-set mutex — so the snapshot is not a single atomic cut across all fields. Publish it however you like — expvar.Publish or a Prometheus collector are a few lines each; the library depends on neither.