kvcache-ai/Mooncake

Fork 0

mirror of https://github.com/kvcache-ai/Mooncake.git synced 2026-11-05 01:17:55 +00:00

[GH-ISSUE #1650] [RFC]: HA OpLog Store Abstraction Layer #484

New issue

Closed

opened 2026-10-29 21:15:19 +00:00 by bsakaguchi · 3 comments

bsakaguchi commented

2026-10-29 21:15:19 +00:00

Owner

Originally created by @duhaode520 on GitHub (Mar 13, 2026).
Original GitHub issue: https://github.com/kvcache-ai/Mooncake/issues/1650

Introduction

The HA (High Availability) system introduced in RFC #1200 and implemented in PR #1515 uses etcd as the sole OpLog storage backend. While etcd provides strong consistency and a built-in Watch mechanism, the tight coupling between all HA components and etcd makes it impossible to swap storage backends, difficult to unit test without a running etcd cluster.

This RFC introduces an abstraction layer that decouples upper-layer components (OpLogManager, OpLogApplier, OpLogReplicator) from the etcd-specific implementation, enabling pluggable storage backends (HDFS, S3, local filesystem, etc.) in the future. The change is interface extraction only — no new backends are introduced. The existing EtcdOpLogStore becomes the first (and currently only) implementation of the new abstract interfaces.

Motivation

The OpLog subsystem is tightly coupled to etcd as the sole storage backend. All four core components directly depend on the concrete EtcdOpLogStore class:

Component	etcd Coupling	Consequence
`OpLogManager`	Holds `shared_ptr<EtcdOpLogStore>`	Cannot swap backend without modifying manager
`OpLogApplier`	Lazy-creates `EtcdOpLogStore` behind `#ifdef STORE_USE_ETCD`
`OpLogWatcher`	575-line monolith: directly calls `EtcdHelper` for Watch/Read, owns its own `EtcdOpLogStore`, duplicates `DeserializeOpLogEntry`	watch/reconnect/deserialization/apply all mixed in one class
`HotStandbyService`	Directly constructs `EtcdOpLogStore` at 4 separate sites	Adding a new backend requires touching every construction site

This coupling means: (1) no alternative storage backend can be used without rewriting all consumers; (2) core replication logic cannot be unit tested without a running etcd cluster;

Goals

Pluggable storage backends: Extract OpLogStore and OpLogChangeNotifier abstract interfaces so new backends (HDFS, S3, etc.) can be added without modifying consumers
Testable without etcd: Introduce MockOpLogStore for fast, deterministic unit tests — enable the 23+ previously-SKIP'd test cases
Single Responsibility: Split the 600-line OpLogWatcher monolith into focused components with clear ownership boundaries
Backend-agnostic serialization: Extract serialization as a shared utility, eliminating duplication
Centralized creation: Factory pattern to replace 4 scattered EtcdOpLogStore construction sites
Zero behavioral change: All existing integration tests must continue to pass

Non-Goals: Implementing new storage backends, changing serialization format, modifying etcd key schema, or changing the HA election mechanism.

Proposal

1. Architecture Comparison: Before vs After

(1) Before: etcd-Coupled Architecture (PR #1515)

graph TB
    subgraph "Primary Node"
        Client1[Client] --> MS1[MasterService]
        MS1 --> OLM1[OpLogManager]
        OLM1 -->|"shared_ptr&lt;EtcdOpLogStore&gt;"| EOS1["EtcdOpLogStore<br/>(concrete class)"]
        EOS1 -->|"EtcdHelper::Put/Create/BatchCreate"| ETCD[(etcd Cluster)]
    end

    subgraph "Standby Node"
        HSS1[HotStandbyService] -->|"new EtcdOpLogStore(...)"| EOS2["EtcdOpLogStore<br/>(directly constructed)"]
        HSS1 -->|"new OpLogWatcher(etcd_endpoints)"| OW1["OpLogWatcher<br/>(575 lines, etcd-coupled)"]
        OW1 -->|"owns EtcdOpLogStore"| EOS3["EtcdOpLogStore<br/>(internally created)"]
        OW1 -->|"EtcdHelper::Watch"| ETCD
        OW1 -->|"EtcdHelper::ReadSince"| ETCD
        OW1 --> OA1[OpLogApplier]
        OA1 -->|"lazy EtcdOpLogStore<br/>#ifdef STORE_USE_ETCD"| EOS4["EtcdOpLogStore<br/>(lazy created)"]
        OA1 --> MDS1[MetadataStore]
    end

    style EOS1 fill:#e74c3c,color:#fff
    style EOS2 fill:#e74c3c,color:#fff
    style EOS3 fill:#e74c3c,color:#fff
    style EOS4 fill:#e74c3c,color:#fff
    style OW1 fill:#e74c3c,color:#fff
    style ETCD fill:#778ca3,color:#fff

(2) After: Abstracted Architecture (This RFC)

graph TB
    subgraph "Primary Node"
        Client2[Client] --> MS2[MasterService]
        MS2 --> OLM2[OpLogManager]
        OLM2 -->|"shared_ptr&lt;OpLogStore&gt;"| IF_OLS["OpLogStore<br/>(abstract interface)"]
        IF_OLS -.->|implements| EOS_P["EtcdOpLogStore"]
        EOS_P --> ETCD2[(etcd Cluster)]
    end

    subgraph "Standby Node"
        HSS2[HotStandbyService] -->|"Factory::Create(ETCD, READER)"| FAC[OpLogStoreFactory]
        FAC -->|creates| IF_OLS2["OpLogStore<br/>(abstract)"]
        IF_OLS2 -.->|implements| EOS_S[EtcdOpLogStore]

        IF_OLS2 -->|"CreateChangeNotifier()"| IF_CN["OpLogChangeNotifier<br/>(abstract interface)"]
        IF_CN -.->|implements| ECN["EtcdOpLogChangeNotifier"]

        HSS2 --> OR["OpLogReplicator<br/>(~80 lines, orchestration only)"]
        OR -->|"Start/Stop"| IF_CN
        IF_CN -->|"on_entry callback"| OR
        OR -->|"ApplyOpLogEntry"| OA2[OpLogApplier]
        OA2 -->|"ReadOpLog (gap fill)"| IF_OLS2
        OA2 --> MDS2[MetadataStore]

        ECN -->|"Watch + ReadSince"| ETCD2
    end

    style IF_OLS fill:#3498db,color:#fff
    style IF_OLS2 fill:#3498db,color:#fff
    style IF_CN fill:#2ecc71,color:#fff
    style FAC fill:#f39c12,color:#fff
    style OR fill:#2ecc71,color:#fff
    style ETCD2 fill:#778ca3,color:#fff

Key differences:

Single factory creation point (orange) replaces 4 scattered constructors
Abstract interfaces (blue/green) decouple all consumers from etcd
OpLogReplicator is a thin orchestrator — all etcd logic lives in EtcdOpLogChangeNotifier
OpLogApplier receives OpLogStore* via dependency injection, no #ifdef

2. Component Dependency Comparison

classDiagram
    class OpLogStore {
        <<interface>>
        +Init() ErrorCode
        +WriteOpLog(entry, sync) ErrorCode
        +ReadOpLog(seq_id, entry) ErrorCode
        +ReadOpLogSince(start, limit, entries) ErrorCode
        +GetLatestSequenceId(seq_id) ErrorCode
        +GetMaxSequenceId(seq_id) ErrorCode
        +UpdateLatestSequenceId(seq_id) ErrorCode
        +RecordSnapshotSequenceId(snap, seq) ErrorCode
        +GetSnapshotSequenceId(snap, seq) ErrorCode
        +CleanupOpLogBefore(seq) ErrorCode
        +CreateChangeNotifier(cluster_id) OpLogChangeNotifier*
    }

    class OpLogChangeNotifier {
        <<interface>>
        +Start(start_seq, on_entry, on_error) ErrorCode
        +Stop() void
        +IsHealthy() bool
    }

    class EtcdOpLogStore {
        +ReadOpLogSinceWithRevision(...)
        -BatchWriteThread()
        -FlushBatch()
        -BuildOpLogKey(seq) string
    }

    class EtcdOpLogChangeNotifier {
        -WatchLoop()
        -HandleWatchEvent(key, value)
        -TryReconnect()
        -SyncMissedEntries()
    }

    class MockOpLogStore {
        +SetWriteError(err)
        +SetReadError(err)
        +Clear()
        +EntryCount() size_t
    }

    class OpLogStoreFactory {
        +Create(type, cluster_id, role)$ OpLogStore*
    }

    class OpLogManager {
        -oplog_store_ : OpLogStore*
        +SetOpLogStore(store)
        +Append(type, key, payload) uint64
        +PersistEntry(entry) ErrorCode
    }

    class OpLogApplier {
        -oplog_store_ : OpLogStore*
        +ApplyOpLogEntry(entry) bool
        +ProcessPendingEntries() size_t
        +TryResolveGapsOnceForPromotion() GapResolveResult
    }

    class OpLogReplicator {
        -notifier_ : OpLogChangeNotifier*
        -applier_ : OpLogApplier*
        +StartFromSequenceId(seq) bool
        +Stop()
        +IsHealthy() bool
    }

    class HotStandbyService {
        +Start() ErrorCode
        +Stop()
        +Promote() ErrorCode
    }

    OpLogStore <|.. EtcdOpLogStore : implements
    OpLogStore <|.. MockOpLogStore : implements
    OpLogChangeNotifier <|.. EtcdOpLogChangeNotifier : implements

    OpLogManager --> OpLogStore : writes via
    OpLogApplier --> OpLogStore : reads missing entries
    OpLogReplicator --> OpLogChangeNotifier : receives entries
    OpLogReplicator --> OpLogApplier : applies entries
    OpLogStore ..> OpLogChangeNotifier : creates via factory method
    OpLogStoreFactory ..> OpLogStore : creates

    HotStandbyService --> OpLogStoreFactory : creates store
    HotStandbyService *-- OpLogReplicator
    HotStandbyService *-- OpLogApplier

3. Core Interfaces

(1) OpLogStore — Storage Abstraction

// oplog_store.h
class OpLogStore {
public:
    virtual ~OpLogStore() = default;
    virtual ErrorCode Init() = 0;

    // Write
    virtual ErrorCode WriteOpLog(const OpLogEntry& entry, bool sync = true) = 0;

    // Read
    virtual ErrorCode ReadOpLog(uint64_t sequence_id, OpLogEntry& entry) = 0;
    virtual ErrorCode ReadOpLogSince(uint64_t start_sequence_id, size_t limit,
                                     std::vector<OpLogEntry>& entries) = 0;

    // Sequence ID management
    virtual ErrorCode GetLatestSequenceId(uint64_t& sequence_id) = 0;
    virtual ErrorCode GetMaxSequenceId(uint64_t& sequence_id) = 0;
    virtual ErrorCode UpdateLatestSequenceId(uint64_t sequence_id) = 0;

    // Snapshot-OpLog coordination
    virtual ErrorCode RecordSnapshotSequenceId(const std::string& snapshot_id,
                                               uint64_t sequence_id) = 0;
    virtual ErrorCode GetSnapshotSequenceId(const std::string& snapshot_id,
                                            uint64_t& sequence_id) = 0;

    // Cleanup
    virtual ErrorCode CleanupOpLogBefore(uint64_t before_sequence_id) = 0;

    // Change notification factory (backend-specific)
    virtual std::unique_ptr<OpLogChangeNotifier> CreateChangeNotifier(
        const std::string& cluster_id) { return nullptr; }
};

Key Design Decisions:

a. ReadOpLogSinceWithRevision is NOT in the interface. The etcd revision is a backend-specific concept for "read-then-watch" consistency. Polling-based backends use sequence_id as the checkpoint. This method remains on EtcdOpLogStore only.

b. Snapshot-related methods are included. OpLog GC depends on snapshot boundaries — CleanupOpLogBefore must not delete entries newer than the latest snapshot's sequence_id, so this mapping belongs in the same backend.

(2) OpLogChangeNotifier — Replication Abstraction

// oplog_change_notifier.h
class OpLogChangeNotifier {
public:
    virtual ~OpLogChangeNotifier() = default;

    using EntryCallback = std::function<void(const OpLogEntry& entry)>;
    using ErrorCallback = std::function<void(ErrorCode error)>;

    virtual ErrorCode Start(uint64_t start_sequence_id,
                            EntryCallback on_entry,
                            ErrorCallback on_error) = 0;
    virtual void Stop() = 0;
    virtual bool IsHealthy() const = 0;
};

Implementations:

Implementation	Mechanism	Latency	Backend Requirement
`EtcdOpLogChangeNotifier`	Push via etcd Watch	Milliseconds	etcd Watch API
(future) `PollingOpLogChangeNotifier`	Pull via `ReadOpLogSince`	Seconds (configurable)	Any `OpLogStore`

(3) OpLogStoreFactory — Centralized Creation

// oplog_store_factory.h
enum class OpLogStoreRole {
    WRITER,  // Primary: enables batch_write + batch_update background threads
    READER,  // Standby: read-only, no background threads
};

enum class OpLogStoreType {
    ETCD,
    // Future: HDFS, S3, LOCAL_FS, ...
};

class OpLogStoreFactory {
public:
    static std::unique_ptr<OpLogStore> Create(
        OpLogStoreType type, const std::string& cluster_id, OpLogStoreRole role);
};

The factory replaces 4 scattered EtcdOpLogStore construction sites. Adding a new backend requires only:

Implement OpLogStore + OpLogChangeNotifier
Add an enum value to OpLogStoreType
Add a case to OpLogStoreFactory::Create

(4) Serialization Extraction

// oplog_serializer.h — backend-agnostic, shared by all implementations
std::string SerializeOpLogEntry(const OpLogEntry& entry);
bool DeserializeOpLogEntry(const std::string& json_str, OpLogEntry& entry);

Previously duplicated as private methods in both EtcdOpLogStore and OpLogWatcher. Now a single shared utility using JSON format with base64-encoded binary payloads.

4. OpLogWatcher Refactoring: Before vs After

This is the most significant structural change. The monolithic OpLogWatcher is split into focused components:

graph LR
    subgraph "Before: OpLogWatcher (575 lines)"
        OW["OpLogWatcher"] -->|"directly calls"| EH[EtcdHelper]
        OW -->|"directly creates"| EOS_old[EtcdOpLogStore]
        OW -->|"manages"| WCB[WatchCallbackContext]
        OW -->|"manages"| WT[watch_thread_]
        OW -->|"reconnection"| RC[TryReconnect]
        OW -->|"deserialization"| DS[DeserializeOpLogEntry]
        OW --> OA_old[OpLogApplier]
    end

    style OW fill:#e74c3c,color:#fff

graph LR
    subgraph "After: Separated Responsibilities"
        OR_new["OpLogReplicator<br/>(~80 lines)"] -->|"delegates to"| CN["OpLogChangeNotifier<br/>(interface)"]
        OR_new --> OA_new[OpLogApplier]

        CN -.->|implements| ECN["EtcdOpLogChangeNotifier<br/>(watch + reconnect + sync)"]
        ECN --> EOS_new[EtcdOpLogStore]
    end

    style OR_new fill:#2ecc71,color:#fff
    style CN fill:#3498db,color:#fff
    style ECN fill:#ff9f43,color:#fff

Aspect	Before (`OpLogWatcher`)	After (split)
Lines of code	~575 lines in one class	~80 (Replicator) + ~300 (Notifier)
etcd coupling	Direct `EtcdHelper` calls + own `EtcdOpLogStore`	All etcd logic in `EtcdOpLogChangeNotifier`
Serialization	Private duplicate `DeserializeOpLogEntry`	Shared `oplog_serializer.h`
Callback safety	Raw `WatchCallbackContext*` managed inline	`ChangeNotifierCallbackContext` encapsulated
Testability	Requires running etcd	Mock `OpLogChangeNotifier` for unit tests
Naming	"Watcher" (etcd-centric)	"Replicator" (backend-agnostic)

5. Upper-Layer Component Changes

Component	Before (PR #1515)	After (This RFC)	Change Type
OpLogManager	`shared_ptr<EtcdOpLogStore>`	`shared_ptr<OpLogStore>`	Type widening
	`SetEtcdOpLogStore()`	`SetOpLogStore()`	Rename
	`PersistEntryToEtcd()`	`PersistEntry()`	Rename
OpLogApplier	Lazy-creates `EtcdOpLogStore` internally	Receives `OpLogStore*` via constructor	Dependency injection
	`#ifdef STORE_USE_ETCD` guards	Runtime `if (oplog_store_)`	Compile-time → runtime
	`GetEtcdOpLogStore()` private method	`oplog_store_` pointer (no lazy init)	Simplification
OpLogReplicator	`OpLogWatcher` (575 LOC)	`OpLogReplicator` (~80 LOC)	Decomposition
	Accepts `etcd_endpoints` string	Accepts `OpLogChangeNotifier` + `OpLogApplier`	Interface injection
HotStandbyService	`new EtcdOpLogStore(...)` directly	`OpLogStoreFactory::Create(ETCD, id, READER)`	Factory pattern
	`#include "etcd_oplog_store.h"`	`#include "oplog_store_factory.h"`	Dependency inversion

6. etcd-Specific Implementation Details

These details are unchanged from PR #1515 and are hidden behind the abstract interfaces. See RFC #1200 and docs/oplog-store-abstraction-architecture.md for full HA flow sequences (startup sync, gap recovery, watch reconnect, promotion).

7. `#ifdef STORE_USE_ETCD` Strategy

Layer	Before	After
`EtcdOpLogStore`, `EtcdOpLogChangeNotifier`	`#ifdef STORE_USE_ETCD`	Retained — these files require etcd
`OpLogStoreFactory`	N/A (didn't exist)	`#ifdef` in `.cpp` for etcd case
`OpLogManager`	N/A	No `#ifdef` — depends on abstract `OpLogStore`
`OpLogApplier`	`#ifdef STORE_USE_ETCD` for lazy init	Removed — runtime `if (oplog_store_)`
`OpLogWatcher` → `OpLogReplicator`	`#ifdef` for etcd watch	Removed — depends on abstract `OpLogChangeNotifier`
`HotStandbyService`	`#include "etcd_oplog_store.h"`	`#include "oplog_store_factory.h"`

Testing Strategy

New Test Infrastructure

MockOpLogStore (tests/hot_standby_ut/mock_oplog_store.h):

In-memory std::map<uint64_t, OpLogEntry> storage
Fault injection: SetWriteError(), SetReadError()
CreateChangeNotifier() returns a mock notifier
Enables fast unit tests without etcd

New and Enabled Tests

Test Suite	Cases	Status Change
`oplog_serializer_test`	11	New — validates JSON + base64 round-trip, edge cases
`oplog_applier_test` (gap resolution)	5	SKIP → Pass — uses MockOpLogStore
`oplog_replicator_test` (event handling)	4	SKIP → Pass — uses mock ChangeNotifier
`oplog_manager_test` (boundary)	3	New — MockOpLogStore integration

File Organization

New Files

File	Description
`include/oplog_store.h`	`OpLogStore` abstract interface
`include/oplog_change_notifier.h`	`OpLogChangeNotifier` abstract interface
`include/oplog_serializer.h`	Backend-agnostic serialization
`include/oplog_store_factory.h`	Factory with role/type config
`include/etcd_oplog_change_notifier.h`	etcd Watch-based notifier
`src/oplog_serializer.cpp`	Serialization implementation
`src/oplog_store_factory.cpp`	Factory implementation
`src/etcd_oplog_change_notifier.cpp`	etcd Watch implementation
`tests/hot_standby_ut/mock_oplog_store.h`	In-memory mock
`tests/hot_standby_ut/oplog_serializer_test.cpp`	Serialization tests

Modified Files

File	Change
`include/etcd_oplog_store.h`	Inherits `OpLogStore`, adds `override`
`include/oplog_manager.h`	`EtcdOpLogStore` → `OpLogStore`
`include/oplog_applier.h`	`EtcdOpLogStore` → `OpLogStore`, DI
`include/oplog_replicator.h`	Was `oplog_watcher.h`, now depends on `OpLogChangeNotifier`
`src/etcd_oplog_store.cpp`	Serialization extracted
`src/oplog_manager.cpp`	`SetOpLogStore()`, `PersistEntry()`
`src/oplog_applier.cpp`	Constructor injection, no `#ifdef`
`src/oplog_replicator.cpp`	~80 lines (was ~575 in `oplog_watcher.cpp`)
`src/hot_standby_service.cpp`	Uses `OpLogStoreFactory`

Future Work

Alternative Storage Backends

Backend	Push/Poll	Key Advantage	Estimated Effort
S3/COS	Poll	Cloud-native, cheap	~600 lines
Local FS	Poll	Zero external dependency	~400 lines

Key Design Points Summary

Minimal interface extraction: OpLogStore mirrors EtcdOpLogStore's public API exactly — no speculative methods
Backend-specific details stay behind the interface: etcd revision, batch transactions, watch callbacks are all internal to the etcd implementation
Factory centralizes creation: 4 scattered construction sites → 1 factory with role-based configuration
Serialization is shared: JSON + base64 format is backend-agnostic, extracted as a utility
Test-first approach: MockOpLogStore and 23+ new/enabled tests added before any interface changes
Zero behavioral change: All existing integration tests pass without modification
Incremental migration: New backends can be added one at a time with no changes to upper-layer code

Before submitting a new issue...

Make sure you already searched for relevant issues and read the documentation

Originally created by @duhaode520 on GitHub (Mar 13, 2026). Original GitHub issue: https://github.com/kvcache-ai/Mooncake/issues/1650 ## Introduction The HA (High Availability) system introduced in [RFC #1200](https://github.com/kvcache-ai/Mooncake/issues/1200) and implemented in PR [#1515](https://github.com/kvcache-ai/Mooncake/pull/1515) uses etcd as the sole OpLog storage backend. While etcd provides strong consistency and a built-in Watch mechanism, the tight coupling between all HA components and etcd makes it impossible to swap storage backends, difficult to unit test without a running etcd cluster. This RFC introduces an **abstraction layer** that decouples upper-layer components (`OpLogManager`, `OpLogApplier`, `OpLogReplicator`) from the etcd-specific implementation, enabling pluggable storage backends (HDFS, S3, local filesystem, etc.) in the future. The change is **interface extraction only** — no new backends are introduced. The existing `EtcdOpLogStore` becomes the first (and currently only) implementation of the new abstract interfaces. ## Motivation The OpLog subsystem is **tightly coupled to etcd** as the sole storage backend. All four core components directly depend on the concrete `EtcdOpLogStore` class: | Component | etcd Coupling | Consequence | |-----------|--------------|-------------| | `OpLogManager` | Holds `shared_ptr<EtcdOpLogStore>` | Cannot swap backend without modifying manager | | `OpLogApplier` | Lazy-creates `EtcdOpLogStore` behind `#ifdef STORE_USE_ETCD` || | `OpLogWatcher` | 575-line monolith: directly calls `EtcdHelper` for Watch/Read, owns its own `EtcdOpLogStore`, duplicates `DeserializeOpLogEntry` | watch/reconnect/deserialization/apply all mixed in one class | | `HotStandbyService` | Directly constructs `EtcdOpLogStore` at 4 separate sites | Adding a new backend requires touching every construction site | This coupling means: (1) no alternative storage backend can be used without rewriting all consumers; (2) core replication logic cannot be unit tested without a running etcd cluster; ## Goals 1. **Pluggable storage backends**: Extract `OpLogStore` and `OpLogChangeNotifier` abstract interfaces so new backends (HDFS, S3, etc.) can be added without modifying consumers 2. **Testable without etcd**: Introduce `MockOpLogStore` for fast, deterministic unit tests — enable the 23+ previously-SKIP'd test cases 3. **Single Responsibility**: Split the 600-line `OpLogWatcher` monolith into focused components with clear ownership boundaries 4. **Backend-agnostic serialization**: Extract serialization as a shared utility, eliminating duplication 5. **Centralized creation**: Factory pattern to replace 4 scattered `EtcdOpLogStore` construction sites 6. **Zero behavioral change**: All existing integration tests must continue to pass **Non-Goals**: Implementing new storage backends, changing serialization format, modifying etcd key schema, or changing the HA election mechanism. ## Proposal ### 1. Architecture Comparison: Before vs After #### (1) Before: etcd-Coupled Architecture (PR #1515) ```mermaid graph TB subgraph "Primary Node" Client1[Client] --> MS1[MasterService] MS1 --> OLM1[OpLogManager] OLM1 -->|"shared_ptr<EtcdOpLogStore>"| EOS1["EtcdOpLogStore (concrete class)"] EOS1 -->|"EtcdHelper::Put/Create/BatchCreate"| ETCD[(etcd Cluster)] end subgraph "Standby Node" HSS1[HotStandbyService] -->|"new EtcdOpLogStore(...)"| EOS2["EtcdOpLogStore (directly constructed)"] HSS1 -->|"new OpLogWatcher(etcd_endpoints)"| OW1["OpLogWatcher (575 lines, etcd-coupled)"] OW1 -->|"owns EtcdOpLogStore"| EOS3["EtcdOpLogStore (internally created)"] OW1 -->|"EtcdHelper::Watch"| ETCD OW1 -->|"EtcdHelper::ReadSince"| ETCD OW1 --> OA1[OpLogApplier] OA1 -->|"lazy EtcdOpLogStore #ifdef STORE_USE_ETCD"| EOS4["EtcdOpLogStore (lazy created)"] OA1 --> MDS1[MetadataStore] end style EOS1 fill:#e74c3c,color:#fff style EOS2 fill:#e74c3c,color:#fff style EOS3 fill:#e74c3c,color:#fff style EOS4 fill:#e74c3c,color:#fff style OW1 fill:#e74c3c,color:#fff style ETCD fill:#778ca3,color:#fff ``` #### (2) After: Abstracted Architecture (This RFC) ```mermaid graph TB subgraph "Primary Node" Client2[Client] --> MS2[MasterService] MS2 --> OLM2[OpLogManager] OLM2 -->|"shared_ptr<OpLogStore>"| IF_OLS["OpLogStore (abstract interface)"] IF_OLS -.->|implements| EOS_P["EtcdOpLogStore"] EOS_P --> ETCD2[(etcd Cluster)] end subgraph "Standby Node" HSS2[HotStandbyService] -->|"Factory::Create(ETCD, READER)"| FAC[OpLogStoreFactory] FAC -->|creates| IF_OLS2["OpLogStore (abstract)"] IF_OLS2 -.->|implements| EOS_S[EtcdOpLogStore] IF_OLS2 -->|"CreateChangeNotifier()"| IF_CN["OpLogChangeNotifier (abstract interface)"] IF_CN -.->|implements| ECN["EtcdOpLogChangeNotifier"] HSS2 --> OR["OpLogReplicator (~80 lines, orchestration only)"] OR -->|"Start/Stop"| IF_CN IF_CN -->|"on_entry callback"| OR OR -->|"ApplyOpLogEntry"| OA2[OpLogApplier] OA2 -->|"ReadOpLog (gap fill)"| IF_OLS2 OA2 --> MDS2[MetadataStore] ECN -->|"Watch + ReadSince"| ETCD2 end style IF_OLS fill:#3498db,color:#fff style IF_OLS2 fill:#3498db,color:#fff style IF_CN fill:#2ecc71,color:#fff style FAC fill:#f39c12,color:#fff style OR fill:#2ecc71,color:#fff style ETCD2 fill:#778ca3,color:#fff ``` **Key differences**: - Single factory creation point (orange) replaces 4 scattered constructors - Abstract interfaces (blue/green) decouple all consumers from etcd - `OpLogReplicator` is a thin orchestrator — all etcd logic lives in `EtcdOpLogChangeNotifier` - `OpLogApplier` receives `OpLogStore*` via dependency injection, no `#ifdef` ### 2. Component Dependency Comparison ```mermaid classDiagram class OpLogStore { <<interface>> +Init() ErrorCode +WriteOpLog(entry, sync) ErrorCode +ReadOpLog(seq_id, entry) ErrorCode +ReadOpLogSince(start, limit, entries) ErrorCode +GetLatestSequenceId(seq_id) ErrorCode +GetMaxSequenceId(seq_id) ErrorCode +UpdateLatestSequenceId(seq_id) ErrorCode +RecordSnapshotSequenceId(snap, seq) ErrorCode +GetSnapshotSequenceId(snap, seq) ErrorCode +CleanupOpLogBefore(seq) ErrorCode +CreateChangeNotifier(cluster_id) OpLogChangeNotifier* } class OpLogChangeNotifier { <<interface>> +Start(start_seq, on_entry, on_error) ErrorCode +Stop() void +IsHealthy() bool } class EtcdOpLogStore { +ReadOpLogSinceWithRevision(...) -BatchWriteThread() -FlushBatch() -BuildOpLogKey(seq) string } class EtcdOpLogChangeNotifier { -WatchLoop() -HandleWatchEvent(key, value) -TryReconnect() -SyncMissedEntries() } class MockOpLogStore { +SetWriteError(err) +SetReadError(err) +Clear() +EntryCount() size_t } class OpLogStoreFactory { +Create(type, cluster_id, role)$ OpLogStore* } class OpLogManager { -oplog_store_ : OpLogStore* +SetOpLogStore(store) +Append(type, key, payload) uint64 +PersistEntry(entry) ErrorCode } class OpLogApplier { -oplog_store_ : OpLogStore* +ApplyOpLogEntry(entry) bool +ProcessPendingEntries() size_t +TryResolveGapsOnceForPromotion() GapResolveResult } class OpLogReplicator { -notifier_ : OpLogChangeNotifier* -applier_ : OpLogApplier* +StartFromSequenceId(seq) bool +Stop() +IsHealthy() bool } class HotStandbyService { +Start() ErrorCode +Stop() +Promote() ErrorCode } OpLogStore <|.. EtcdOpLogStore : implements OpLogStore <|.. MockOpLogStore : implements OpLogChangeNotifier <|.. EtcdOpLogChangeNotifier : implements OpLogManager --> OpLogStore : writes via OpLogApplier --> OpLogStore : reads missing entries OpLogReplicator --> OpLogChangeNotifier : receives entries OpLogReplicator --> OpLogApplier : applies entries OpLogStore ..> OpLogChangeNotifier : creates via factory method OpLogStoreFactory ..> OpLogStore : creates HotStandbyService --> OpLogStoreFactory : creates store HotStandbyService *-- OpLogReplicator HotStandbyService *-- OpLogApplier ``` ### 3. Core Interfaces #### (1) OpLogStore — Storage Abstraction ```cpp // oplog_store.h class OpLogStore { public: virtual ~OpLogStore() = default; virtual ErrorCode Init() = 0; // Write virtual ErrorCode WriteOpLog(const OpLogEntry& entry, bool sync = true) = 0; // Read virtual ErrorCode ReadOpLog(uint64_t sequence_id, OpLogEntry& entry) = 0; virtual ErrorCode ReadOpLogSince(uint64_t start_sequence_id, size_t limit, std::vector<OpLogEntry>& entries) = 0; // Sequence ID management virtual ErrorCode GetLatestSequenceId(uint64_t& sequence_id) = 0; virtual ErrorCode GetMaxSequenceId(uint64_t& sequence_id) = 0; virtual ErrorCode UpdateLatestSequenceId(uint64_t sequence_id) = 0; // Snapshot-OpLog coordination virtual ErrorCode RecordSnapshotSequenceId(const std::string& snapshot_id, uint64_t sequence_id) = 0; virtual ErrorCode GetSnapshotSequenceId(const std::string& snapshot_id, uint64_t& sequence_id) = 0; // Cleanup virtual ErrorCode CleanupOpLogBefore(uint64_t before_sequence_id) = 0; // Change notification factory (backend-specific) virtual std::unique_ptr<OpLogChangeNotifier> CreateChangeNotifier( const std::string& cluster_id) { return nullptr; } }; ``` **Key Design Decisions**: a. **`ReadOpLogSinceWithRevision` is NOT in the interface.** The etcd revision is a backend-specific concept for "read-then-watch" consistency. Polling-based backends use `sequence_id` as the checkpoint. This method remains on `EtcdOpLogStore` only. b. **Snapshot-related methods are included.** OpLog GC depends on snapshot boundaries — `CleanupOpLogBefore` must not delete entries newer than the latest snapshot's `sequence_id`, so this mapping belongs in the same backend. #### (2) OpLogChangeNotifier — Replication Abstraction ```cpp // oplog_change_notifier.h class OpLogChangeNotifier { public: virtual ~OpLogChangeNotifier() = default; using EntryCallback = std::function<void(const OpLogEntry& entry)>; using ErrorCallback = std::function<void(ErrorCode error)>; virtual ErrorCode Start(uint64_t start_sequence_id, EntryCallback on_entry, ErrorCallback on_error) = 0; virtual void Stop() = 0; virtual bool IsHealthy() const = 0; }; ``` **Implementations**: | Implementation | Mechanism | Latency | Backend Requirement | |---------------|-----------|---------|-------------------| | `EtcdOpLogChangeNotifier` | Push via etcd Watch | Milliseconds | etcd Watch API | | (future) `PollingOpLogChangeNotifier` | Pull via `ReadOpLogSince` | Seconds (configurable) | Any `OpLogStore` | #### (3) OpLogStoreFactory — Centralized Creation ```cpp // oplog_store_factory.h enum class OpLogStoreRole { WRITER, // Primary: enables batch_write + batch_update background threads READER, // Standby: read-only, no background threads }; enum class OpLogStoreType { ETCD, // Future: HDFS, S3, LOCAL_FS, ... }; class OpLogStoreFactory { public: static std::unique_ptr<OpLogStore> Create( OpLogStoreType type, const std::string& cluster_id, OpLogStoreRole role); }; ``` The factory replaces 4 scattered `EtcdOpLogStore` construction sites. Adding a new backend requires only: 1. Implement `OpLogStore` + `OpLogChangeNotifier` 2. Add an enum value to `OpLogStoreType` 3. Add a case to `OpLogStoreFactory::Create` #### (4) Serialization Extraction ```cpp // oplog_serializer.h — backend-agnostic, shared by all implementations std::string SerializeOpLogEntry(const OpLogEntry& entry); bool DeserializeOpLogEntry(const std::string& json_str, OpLogEntry& entry); ``` Previously duplicated as private methods in both `EtcdOpLogStore` and `OpLogWatcher`. Now a single shared utility using JSON format with base64-encoded binary payloads. ### 4. OpLogWatcher Refactoring: Before vs After This is the most significant structural change. The monolithic `OpLogWatcher` is split into focused components: ```mermaid graph LR subgraph "Before: OpLogWatcher (575 lines)" OW["OpLogWatcher"] -->|"directly calls"| EH[EtcdHelper] OW -->|"directly creates"| EOS_old[EtcdOpLogStore] OW -->|"manages"| WCB[WatchCallbackContext] OW -->|"manages"| WT[watch_thread_] OW -->|"reconnection"| RC[TryReconnect] OW -->|"deserialization"| DS[DeserializeOpLogEntry] OW --> OA_old[OpLogApplier] end style OW fill:#e74c3c,color:#fff ``` ```mermaid graph LR subgraph "After: Separated Responsibilities" OR_new["OpLogReplicator (~80 lines)"] -->|"delegates to"| CN["OpLogChangeNotifier (interface)"] OR_new --> OA_new[OpLogApplier] CN -.->|implements| ECN["EtcdOpLogChangeNotifier (watch + reconnect + sync)"] ECN --> EOS_new[EtcdOpLogStore] end style OR_new fill:#2ecc71,color:#fff style CN fill:#3498db,color:#fff style ECN fill:#ff9f43,color:#fff ``` | Aspect | Before (`OpLogWatcher`) | After (split) | |--------|------------------------|---------------| | Lines of code | ~575 lines in one class | ~80 (Replicator) + ~300 (Notifier) | | etcd coupling | Direct `EtcdHelper` calls + own `EtcdOpLogStore` | All etcd logic in `EtcdOpLogChangeNotifier` | | Serialization | Private duplicate `DeserializeOpLogEntry` | Shared `oplog_serializer.h` | | Callback safety | Raw `WatchCallbackContext*` managed inline | `ChangeNotifierCallbackContext` encapsulated | | Testability | Requires running etcd | Mock `OpLogChangeNotifier` for unit tests | | Naming | "Watcher" (etcd-centric) | "Replicator" (backend-agnostic) | ### 5. Upper-Layer Component Changes | Component | Before (PR #1515) | After (This RFC) | Change Type | |-----------|-------------------|-------------------|-------------| | **OpLogManager** | `shared_ptr<EtcdOpLogStore>` | `shared_ptr<OpLogStore>` | Type widening | | | `SetEtcdOpLogStore()` | `SetOpLogStore()` | Rename | | | `PersistEntryToEtcd()` | `PersistEntry()` | Rename | | **OpLogApplier** | Lazy-creates `EtcdOpLogStore` internally | Receives `OpLogStore*` via constructor | Dependency injection | | | `#ifdef STORE_USE_ETCD` guards | Runtime `if (oplog_store_)` | Compile-time → runtime | | | `GetEtcdOpLogStore()` private method | `oplog_store_` pointer (no lazy init) | Simplification | | **OpLogReplicator** | `OpLogWatcher` (575 LOC) | `OpLogReplicator` (~80 LOC) | Decomposition | | | Accepts `etcd_endpoints` string | Accepts `OpLogChangeNotifier*` + `OpLogApplier*` | Interface injection | | **HotStandbyService** | `new EtcdOpLogStore(...)` directly | `OpLogStoreFactory::Create(ETCD, id, READER)` | Factory pattern | | | `#include "etcd_oplog_store.h"` | `#include "oplog_store_factory.h"` | Dependency inversion | ### 6. etcd-Specific Implementation Details These details are **unchanged from PR #1515** and are hidden behind the abstract interfaces. See [RFC #1200](https://github.com/kvcache-ai/Mooncake/issues/1200) and `docs/oplog-store-abstraction-architecture.md` for full HA flow sequences (startup sync, gap recovery, watch reconnect, promotion). ### 7. `#ifdef STORE_USE_ETCD` Strategy | Layer | Before | After | |-------|--------|-------| | `EtcdOpLogStore`, `EtcdOpLogChangeNotifier` | `#ifdef STORE_USE_ETCD` | Retained — these files require etcd | | `OpLogStoreFactory` | N/A (didn't exist) | `#ifdef` in `.cpp` for etcd case | | `OpLogManager` | N/A | No `#ifdef` — depends on abstract `OpLogStore` | | `OpLogApplier` | `#ifdef STORE_USE_ETCD` for lazy init | Removed — runtime `if (oplog_store_)` | | `OpLogWatcher` → `OpLogReplicator` | `#ifdef` for etcd watch | Removed — depends on abstract `OpLogChangeNotifier` | | `HotStandbyService` | `#include "etcd_oplog_store.h"` | `#include "oplog_store_factory.h"` | ## Testing Strategy ### New Test Infrastructure **`MockOpLogStore`** (`tests/hot_standby_ut/mock_oplog_store.h`): - In-memory `std::map<uint64_t, OpLogEntry>` storage - Fault injection: `SetWriteError()`, `SetReadError()` - `CreateChangeNotifier()` returns a mock notifier - Enables fast unit tests without etcd ### New and Enabled Tests | Test Suite | Cases | Status Change | |------------|-------|--------------| | `oplog_serializer_test` | 11 | **New** — validates JSON + base64 round-trip, edge cases | | `oplog_applier_test` (gap resolution) | 5 | **SKIP → Pass** — uses MockOpLogStore | | `oplog_replicator_test` (event handling) | 4 | **SKIP → Pass** — uses mock ChangeNotifier | | `oplog_manager_test` (boundary) | 3 | **New** — MockOpLogStore integration | ## File Organization ### New Files | File | Description | |------|-------------| | `include/oplog_store.h` | `OpLogStore` abstract interface | | `include/oplog_change_notifier.h` | `OpLogChangeNotifier` abstract interface | | `include/oplog_serializer.h` | Backend-agnostic serialization | | `include/oplog_store_factory.h` | Factory with role/type config | | `include/etcd_oplog_change_notifier.h` | etcd Watch-based notifier | | `src/oplog_serializer.cpp` | Serialization implementation | | `src/oplog_store_factory.cpp` | Factory implementation | | `src/etcd_oplog_change_notifier.cpp` | etcd Watch implementation | | `tests/hot_standby_ut/mock_oplog_store.h` | In-memory mock | | `tests/hot_standby_ut/oplog_serializer_test.cpp` | Serialization tests | ### Modified Files | File | Change | |------|--------| | `include/etcd_oplog_store.h` | Inherits `OpLogStore`, adds `override` | | `include/oplog_manager.h` | `EtcdOpLogStore` → `OpLogStore` | | `include/oplog_applier.h` | `EtcdOpLogStore` → `OpLogStore`, DI | | `include/oplog_replicator.h` | Was `oplog_watcher.h`, now depends on `OpLogChangeNotifier` | | `src/etcd_oplog_store.cpp` | Serialization extracted | | `src/oplog_manager.cpp` | `SetOpLogStore()`, `PersistEntry()` | | `src/oplog_applier.cpp` | Constructor injection, no `#ifdef` | | `src/oplog_replicator.cpp` | ~80 lines (was ~575 in `oplog_watcher.cpp`) | | `src/hot_standby_service.cpp` | Uses `OpLogStoreFactory` | ## Future Work ### Alternative Storage Backends | Backend | Push/Poll | Key Advantage | Estimated Effort | |---------|-----------|---------------|-----------------| | S3/COS | Poll | Cloud-native, cheap | ~600 lines | | Local FS | Poll | Zero external dependency | ~400 lines | ## Key Design Points Summary 1. **Minimal interface extraction**: `OpLogStore` mirrors `EtcdOpLogStore`'s public API exactly — no speculative methods 2. **Backend-specific details stay behind the interface**: etcd revision, batch transactions, watch callbacks are all internal to the etcd implementation 3. **Factory centralizes creation**: 4 scattered construction sites → 1 factory with role-based configuration 4. **Serialization is shared**: JSON + base64 format is backend-agnostic, extracted as a utility 5. **Test-first approach**: MockOpLogStore and 23+ new/enabled tests added before any interface changes 6. **Zero behavioral change**: All existing integration tests pass without modification 7. **Incremental migration**: New backends can be added one at a time with no changes to upper-layer code ### Before submitting a new issue... - [x] Make sure you already searched for relevant issues and read the [documentation](https://kvcache-ai.github.io/Mooncake/)

bsakaguchi

2026-10-29 21:15:19 +00:00

closed this issue
added the
stale

auto-closed
labels

bsakaguchi commented

2026-10-29 21:15:20 +00:00

Author

Owner

@duhaode520 commented on GitHub (Mar 13, 2026):

@00fish0 @Libotry, RFC about pluggable abstraction design for Oplog storage backend is available, do you have time to review this? Looking forward to your suggestion.

@duhaode520 commented on GitHub (Mar 13, 2026): @00fish0 @Libotry, RFC about pluggable abstraction design for Oplog storage backend is available, do you have time to review this? Looking forward to your suggestion.

bsakaguchi commented

2026-10-29 21:15:20 +00:00

Author

Owner

@github-actions[bot] commented on GitHub (Jun 12, 2026):

This issue has had no activity for 90 days and will be closed in 7 days if there is no further activity. Please comment or react if it should stay open.

@github-actions[bot] commented on GitHub (Jun 12, 2026): This issue has had no activity for 90 days and will be closed in 7 days if there is no further activity. Please comment or react if it should stay open.

bsakaguchi commented

2026-10-29 21:15:20 +00:00

Author

Owner

@github-actions[bot] commented on GitHub (Jun 20, 2026):

Closing due to 3 months of inactivity. If this is still relevant, please comment and we can reopen.

@github-actions[bot] commented on GitHub (Jun 20, 2026): Closing due to 3 months of inactivity. If this is still relevant, please comment and we can reopen.