Oxia Federation

[mattisonchao]

Motivation

In systems like Apache Pulsar, the architecture is split into a control plane (managing topic metadata, policies, tenant configs) and a data plane (serving actual message traffic). The data plane needs fast, local access to control plane metadata — but it should not be writing to it.

Today, Oxia replication is confined to a single cluster. If the data plane wants to read control plane metadata, it must make remote calls to the control plane Oxia cluster, adding latency and creating a hard dependency. If the control plane is temporarily unavailable, the data plane loses access to its configuration.

The Core Use Case

We want control plane → data plane data syncing, where:

• A control plane Oxia namespace is the source of truth for configs, policies, and metadata
• One or more data plane Oxia namespaces act as read-only caches that mirror the control plane data locally
• Data flows one-directional: control plane → data plane (the data plane never writes back)
• The data plane serves reads with low latency from its local copy, with eventual consistency

┌──────────────────────────────────────────────────────────────┐
│                     CONTROL PLANE                            │
│                                                              │
│   Oxia Cluster (namespace: "control")                        │
│   ┌────────────────────────────────────────────────┐         │
│   │  Shard 0: policies, tenant configs, ...        │         │
│   │  Shard 1: topic metadata, subscriptions, ...   │         │
│   │  Shard 2: ...                                  │         │
│   └────────┬───────────────────────┬───────────────┘         │
│            │                       │                         │
└────────────┼───────────────────────┼─────────────────────────┘
             │ observer stream       │ observer stream
             ▼                       ▼
┌────────────────────────┐ ┌────────────────────────┐
│     DATA PLANE A       │ │     DATA PLANE B       │
│                        │ │                        │
│  Oxia (ns: "dp-a")     │ │  Oxia (ns: "dp-b")     │
│  ┌──────────────────┐  │ │  ┌──────────────────┐  │
│  │ Shard 0 (cache)  │  │ │  │ Shard 0 (cache)  │  │
│  │ Shard 1 (cache)  │  │ │  │ Shard 1 (cache)  │  │
│  │ read-only serves │  │ │  │ read-only serves │  │
│  └──────────────────┘  │ │  └──────────────────┘  │
│                        │ │                        │
│  + local data plane    │ │  + local data plane    │
│    Oxia namespaces     │ │    Oxia namespaces     │
└────────────────────────┘ └────────────────────────┘

This is different from geo-DR or active-passive failover. The data plane clusters are intentionally read-only mirrors — they are caches, not standby replicas waiting for promotion.

Core Idea

We already have the observer follower concept in the replication protocol (AddFollowerRequest.observer = true, replication.proto#L129), originally designed for shard splitting. An observer:

• Receives the full snapshot + WAL stream from the shard leader via FollowerCursor
• Does not participate in the replication quorum (no impact on write latency or commit offset)
• Is completely transparent to the primary cluster's consistency guarantees

The proposal: generalize this mechanism so that nodes in a remote Oxia cluster (a different namespace/coordinator domain) can subscribe as observers to a source cluster's shard leaders, creating Oxia Cluster Federation.

Architecture

Replication Flow

Control Plane Leader                    Data Plane Observer
     │                                       │
     │  1. FederationController watches      │
     │     control plane shard assignments   │
     │                                       │
     │  2. AddFollower(observer=true)        │
     │──────────────────────────────────────>│
     │                                       │
     │  3. SendSnapshot (initial sync)       │
     │  ┌──────────────────────────────────>│
     │  │   chunk 1 ... chunk N              │
     │  └──────────────────────────────────>│
     │                              ack ────>│
     │                                       │
     │  4. Replicate (continuous WAL tail)   │
     │  ┌──────────────────────────────────>│
     │  │   Append{entry, commit_offset}     │
     │  │   Append{entry, commit_offset}     │
     │  └──────────────────────────────────>│
     │                      Ack{offset} ───>│
     │                                       │
     │                                  Data plane clients
     │                                  read from local cache
     │                                       │

Leader Failover Handling

  Time ─────────────────────────────────────────────────>

  Control Plane:
  ┌──────────────────┐         ┌──────────────────┐
  │ Leader (node-1)  │ ──X──> │ Leader (node-2)   │  (new term)
  └──────────────────┘         └──────────────────┘
         │                            │
         │ observer stream breaks     │ Federation Controller
         │                            │ re-attaches observer
         ▼                            ▼
  ┌────────────────────────────────────────────────┐
  │              Data Plane Cluster                │
  │  Observer detects stream break, serves stale   │
  │  data until Federation Controller re-attaches  │
  │  from new leader (snapshot-if-needed)           │
  └────────────────────────────────────────────────┘

A key property: during control plane failover, the data plane continues serving reads from its local cache. The data may be briefly stale, but availability is preserved.

Design Details

1. Federation Controller

A new component that runs in the data plane cluster (or as a sidecar):

• Connects to the control plane cluster and watches shard assignments (via PushShardAssignments)
• For each control plane shard, registers an observer FollowerCursor from the control plane leader to a designated data plane node
• Handles control plane leader failovers: re-attaches the observer to the new leader
• Reports federation lag metrics

2. Observer Follower — What We Already Have

Component	Status	Notes
AddFollowerRequest.observer	Exists	Proto field already defined
FollowerCursor snapshot transfer	Exists	Sends full DB snapshot to new/behind followers
FollowerCursor WAL tailing	Exists	Streams Append entries continuously
QuorumAckTracker	Needs change	Must skip observer cursors in quorum calculation
Cross-cluster gRPC connectivity	New	Need to establish gRPC connections between clusters

3. QuorumAckTracker Changes

Currently, NewCursorAcker enforces cursorIdxGenerator < replicationFactor - 1. Observer cursors must be tracked separately so they:

• Do not count toward requiredAcks
• Do not block commit offset advancement
• Do report their ack offset for lag monitoring

// Proposed: add a non-voting observer cursor
func (q *quorumAckTracker) NewObserverCursorAcker() ObserverCursorAcker {
    // No quorum index assigned, acks are tracked for monitoring only
}

4. Cache Serving Status on Data Plane

The data plane observer shards serve read-only traffic with eventual consistency:

• ServingStatus_OBSERVER — a new status indicating the shard can serve reads but not writes
• Data plane clients get routed to observer shards for all read operations
• Writes are rejected — the data plane is a cache, not a writable replica
• The data plane can also have its own local namespaces for data-plane-specific state (these are normal read-write Oxia namespaces, independent of federation)

5. Cache Invalidation and Freshness

Since the data plane is a cache, we need to consider:

• Lag monitoring: oxia_federation_lag_offset and oxia_federation_lag_seconds metrics
• Freshness guarantees: Data plane clients can optionally specify a max acceptable lag; if the local cache is too stale, the read can be rejected or forwarded to the control plane
• Version tracking: Each record's ModificationOffset is available to clients, so they can detect staleness

What Needs to Be Built

Phase 1: Core Federation (MVP)

• FederationController component that watches source cluster shard assignments
• ObserverCursorAcker in QuorumAckTracker (non-voting)
• Cross-cluster gRPC connection management (TLS, authentication)
• Federation lag metrics (oxia_federation_lag_offset, oxia_federation_lag_seconds)

Phase 2: Cache Serving

• ServingStatus_OBSERVER status
• Client routing to observer shards for reads (reject writes)
• Optional freshness-aware reads (max lag threshold)

Phase 3: Operational Tooling

• Admin CLI for managing federation links (oxia admin federation add/remove/status)
• Graceful federation teardown (stop replication without data loss)
• Federation health dashboard

Key Advantages

1. Minimal protocol changes: The observer flag and FollowerCursor machinery already exist. We're extending, not rewriting.
2. No impact on control plane performance: Observer acks don't affect commit offset or write latency.
3. Data plane resilience: If the control plane is temporarily unavailable, the data plane continues serving reads from its local cache.
4. Reuses battle-tested replication: Snapshot transfer, WAL tailing, CRC validation, backoff/retry — all proven in production.
5. Clean separation of concerns: The data plane is explicitly a read-only cache; there's no ambiguity about write ownership.

Open Questions

1. Shard topology coupling: Must the data plane have the same shard count and hash ranges as the control plane? (Likely yes for MVP, but remapping would be nice later.)
2. Selective replication: Should we support replicating only a subset of keys (e.g., by key prefix)? This would let the data plane cache only what it needs.
3. Notification forwarding: Should the data plane also serve notifications for control plane changes? This would let data plane services react to config changes without connecting to the control plane.
4. Multi-source federation: Can a data plane namespace observe from multiple control plane namespaces, or is it strictly 1:1?
5. Session handling: Should the data plane support sessions for cached data, or is sessionless read-only access sufficient?

Related

• #909 — Shard Splitting Discussion — Original observer follower design for shard splits