Add background FSM compaction

Author: bootjp

adapter/distribution_server.go

@@ -21,6 +21,7 @@ type DistributionServer struct {
 	engine      *distribution.Engine
 	catalog     *distribution.CatalogStore
 	coordinator kv.Coordinator
+	readTracker *kv.ActiveTimestampTracker
 	reloadRetry struct {
 		attempts int
 		interval time.Duration
@@ -39,6 +40,12 @@ func WithDistributionCoordinator(coordinator kv.Coordinator) DistributionServerO
 	}
 }
 
+func WithDistributionActiveTimestampTracker(tracker *kv.ActiveTimestampTracker) DistributionServerOption {
+	return func(s *DistributionServer) {
+		s.readTracker = tracker
+	}
+}
+
 // WithCatalogReloadRetryPolicy configures the retry policy used after split
 // commit when waiting for the local catalog snapshot to become visible.
 func WithCatalogReloadRetryPolicy(attempts int, interval time.Duration) DistributionServerOption {
@@ -140,6 +147,8 @@ func (s *DistributionServer) SplitRange(ctx context.Context, req *pb.SplitRangeR
 	if err != nil {
 		return nil, err
 	}
+	readPin := s.pinReadTS(snapshot.ReadTS)
+	defer readPin.Release()
 	if err := validateExpectedCatalogVersion(snapshot.Version, req.GetExpectedCatalogVersion()); err != nil {
 		return nil, err
 	}
@@ -175,6 +184,13 @@ func (s *DistributionServer) SplitRange(ctx context.Context, req *pb.SplitRangeR
 	}, nil
 }
 
+func (s *DistributionServer) pinReadTS(ts uint64) *kv.ActiveTimestampToken {
+	if s == nil || s.readTracker == nil {
+		return nil
+	}
+	return s.readTracker.Pin(ts)
+}
+
 func (s *DistributionServer) verifyCatalogLeader() error {
 	if s.coordinator == nil {
 		return grpcStatusError(codes.FailedPrecondition, errDistributionCoordinatorRequired.Error())

adapter/grpc.go

@@ -12,6 +12,8 @@ import (
 	"github.com/bootjp/elastickv/store"
 	"github.com/cockroachdb/errors"
 	"github.com/spaolacci/murmur3"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
 )
 
 var _ pb.RawKVServer = (*GRPCServer)(nil)
@@ -89,6 +91,9 @@ func (r *GRPCServer) RawGet(ctx context.Context, req *pb.RawGetRequest) (*pb.Raw
 	if errors.Is(err, store.ErrKeyNotFound) {
 		return &pb.RawGetResponse{Value: nil, Exists: false}, nil
 	}
+	if errors.Is(err, store.ErrReadTSCompacted) {
+		return nil, errors.WithStack(status.Error(codes.FailedPrecondition, store.ErrReadTSCompacted.Error()))
+	}
 	if err != nil {
 		return nil, errors.WithStack(err)
 	}
@@ -135,6 +140,9 @@ func (r *GRPCServer) RawScanAt(ctx context.Context, req *pb.RawScanAtRequest) (*
 		res, err = r.store.ScanAt(ctx, req.StartKey, req.EndKey, limit, readTS)
 	}
 	if err != nil {
+		if errors.Is(err, store.ErrReadTSCompacted) {
+			return &pb.RawScanAtResponse{Kv: nil}, errors.WithStack(status.Error(codes.FailedPrecondition, store.ErrReadTSCompacted.Error()))
+		}
 		return &pb.RawScanAtResponse{Kv: nil}, errors.WithStack(err)
 	}
 

adapter/redis.go

@@ -243,6 +243,7 @@ type RedisServer struct {
 	listen          net.Listener
 	store           store.MVCCStore
 	coordinator     kv.Coordinator
+	readTracker     *kv.ActiveTimestampTracker
 	redisTranscoder *redisTranscoder
 	pubsub          *redisPubSub
 	scriptMu        sync.RWMutex
@@ -258,6 +259,14 @@ type RedisServer struct {
 	route map[string]func(conn redcon.Conn, cmd redcon.Command)
 }
 
+type RedisServerOption func(*RedisServer)
+
+func WithRedisActiveTimestampTracker(tracker *kv.ActiveTimestampTracker) RedisServerOption {
+	return func(r *RedisServer) {
+		r.readTracker = tracker
+	}
+}
+
 type connState struct {
 	inTxn bool
 	queue []redcon.Command
@@ -283,7 +292,7 @@ type redisResult struct {
 	err     error
 }
 
-func NewRedisServer(listen net.Listener, redisAddr string, store store.MVCCStore, coordinate kv.Coordinator, leaderRedis map[raft.ServerAddress]string, relay *RedisPubSubRelay) *RedisServer {
+func NewRedisServer(listen net.Listener, redisAddr string, store store.MVCCStore, coordinate kv.Coordinator, leaderRedis map[raft.ServerAddress]string, relay *RedisPubSubRelay, opts ...RedisServerOption) *RedisServer {
 	if relay == nil {
 		relay = NewRedisPubSubRelay()
 	}
@@ -383,6 +392,11 @@ func NewRedisServer(listen net.Listener, redisAddr string, store store.MVCCStore
 		cmdZRevRangeByScore: r.zrevrangebyscore,
 		cmdZScore:           r.zscore,
 	}
+	for _, opt := range opts {
+		if opt != nil {
+			opt(r)
+		}
+	}
 
 	return r
 }
@@ -402,6 +416,13 @@ func (r *RedisServer) readTS() uint64 {
 	return snapshotTS(r.coordinator.Clock(), r.store)
 }
 
+func (r *RedisServer) pinReadTS(ts uint64) *kv.ActiveTimestampToken {
+	if r == nil || r.readTracker == nil {
+		return nil
+	}
+	return r.readTracker.Pin(ts)
+}
+
 func (r *RedisServer) Run() error {
 	err := redcon.Serve(r.listen,
 		func(conn redcon.Conn, cmd redcon.Command) {
@@ -1905,6 +1926,8 @@ func (r *RedisServer) runTransaction(queue []redcon.Command) ([]redisResult, err
 	if err != nil {
 		return nil, err
 	}
+	readPin := r.pinReadTS(startTS)
+	defer readPin.Release()
 
 	ctx := &txnContext{
 		server:     r,

adapter/redis_lua.go

@@ -118,6 +118,7 @@ func (r *RedisServer) runLuaScript(conn redcon.Conn, script string, evalArgs [][
 	var reply luaReply
 	err = r.retryRedisWrite(ctx, func() error {
 		scriptCtx := newLuaScriptContext(r)
+		defer scriptCtx.Close()
 		state := newRedisLuaState()
 		defer state.Close()
 		state.SetContext(ctx)
@@ -847,6 +848,7 @@ func (r *RedisServer) execLuaCompat(conn redcon.Conn, command string, args [][]b
 	var reply luaReply
 	err := r.retryRedisWrite(ctx, func() error {
 		scriptCtx := newLuaScriptContext(r)
+		defer scriptCtx.Close()
 		nextReply, err := scriptCtx.exec(command, stringArgs)
 		if err != nil {
 			return err

adapter/redis_lua_context.go

@@ -16,6 +16,7 @@ import (
 type luaScriptContext struct {
 	server  *RedisServer
 	startTS uint64
+	readPin *kv.ActiveTimestampToken
 
 	touched map[string]struct{}
 	deleted map[string]bool
@@ -161,9 +162,11 @@ var luaRenameHandlers = map[redisValueType]luaRenameHandler{
 }
 
 func newLuaScriptContext(server *RedisServer) *luaScriptContext {
+	startTS := server.readTS()
 	return &luaScriptContext{
 		server:  server,
-		startTS: server.readTS(),
+		startTS: startTS,
+		readPin: server.pinReadTS(startTS),
 		touched: map[string]struct{}{},
 		deleted: map[string]bool{},
 		strings: map[string]*luaStringState{},
@@ -176,6 +179,13 @@ func newLuaScriptContext(server *RedisServer) *luaScriptContext {
 	}
 }
 
+func (c *luaScriptContext) Close() {
+	if c == nil || c.readPin == nil {
+		return
+	}
+	c.readPin.Release()
+}
+
 func (c *luaScriptContext) exec(command string, args []string) (luaReply, error) {
 	if handler, ok := luaCommandHandlers[command]; ok {
 		return handler(c, args)

cmd/server/demo.go

@@ -335,7 +335,7 @@ func setupGRPC(r *raft.Raft, st store.MVCCStore, tm *transport.Manager, coordina
 	return s, gs
 }
 
-func setupRedis(ctx context.Context, lc net.ListenConfig, st store.MVCCStore, coordinator *kv.Coordinate, addr, redisAddr, raftRedisMapStr string, relay *adapter.RedisPubSubRelay) (*adapter.RedisServer, error) {
+func setupRedis(ctx context.Context, lc net.ListenConfig, st store.MVCCStore, coordinator *kv.Coordinate, addr, redisAddr, raftRedisMapStr string, relay *adapter.RedisPubSubRelay, readTracker *kv.ActiveTimestampTracker) (*adapter.RedisServer, error) {
 	leaderRedis := make(map[raft.ServerAddress]string)
 	if raftRedisMapStr != "" {
 		parts := strings.SplitSeq(raftRedisMapStr, ",")
@@ -354,7 +354,7 @@ func setupRedis(ctx context.Context, lc net.ListenConfig, st store.MVCCStore, co
 		return nil, errors.WithStack(err)
 	}
 	routedStore := kv.NewLeaderRoutedStore(st, coordinator)
-	return adapter.NewRedisServer(l, redisAddr, routedStore, coordinator, leaderRedis, relay), nil
+	return adapter.NewRedisServer(l, redisAddr, routedStore, coordinator, leaderRedis, relay, adapter.WithRedisActiveTimestampTracker(readTracker)), nil
 }
 
 func run(ctx context.Context, eg *errgroup.Group, cfg config) error {
@@ -369,6 +369,7 @@ func run(ctx context.Context, eg *errgroup.Group, cfg config) error {
 
 	st := store.NewMVCCStore()
 	fsm := kv.NewKvFSM(st)
+	readTracker := kv.NewActiveTimestampTracker()
 
 	// Config
 	c := raft.DefaultConfig()
@@ -402,9 +403,18 @@ func run(ctx context.Context, eg *errgroup.Group, cfg config) error {
 		distEngine,
 		distCatalog,
 		adapter.WithDistributionCoordinator(coordinator),
+		adapter.WithDistributionActiveTimestampTracker(readTracker),
 	)
 	metricsRegistry := monitoring.NewRegistry(cfg.raftID, cfg.address)
 	metricsRegistry.RaftObserver().Start(ctx, []monitoring.RaftRuntime{{GroupID: 1, Raft: r}}, raftObserveInterval)
+	compactor := kv.NewFSMCompactor(
+		[]kv.FSMCompactRuntime{{
+			GroupID: 1,
+			Raft:    r,
+			Store:   st,
+		}},
+		kv.WithFSMCompactorActiveTimestampTracker(readTracker),
+	)
 	relay := adapter.NewRedisPubSubRelay()
 
 	s, grpcSvc := setupGRPC(r, st, tm, coordinator, distServer, relay)
@@ -417,7 +427,7 @@ func run(ctx context.Context, eg *errgroup.Group, cfg config) error {
 		_ = grpcSock.Close()
 	})
 
-	rd, err := setupRedis(ctx, lc, st, coordinator, cfg.address, cfg.redisAddress, cfg.raftRedisMap, relay)
+	rd, err := setupRedis(ctx, lc, st, coordinator, cfg.address, cfg.redisAddress, cfg.raftRedisMap, relay, readTracker)
 	if err != nil {
 		return err
 	}
@@ -444,6 +454,7 @@ func run(ctx context.Context, eg *errgroup.Group, cfg config) error {
 	}
 
 	eg.Go(catalogWatcherTask(ctx, distCatalog, distEngine))
+	eg.Go(func() error { return compactor.Run(ctx) })
 	eg.Go(grpcShutdownTask(ctx, s, grpcSock, cfg.address, grpcSvc))
 	eg.Go(grpcServeTask(s, grpcSock, cfg.address))
 	eg.Go(redisShutdownTask(ctx, rd, cfg.redisAddress))

docs/fsm_compaction_design.md

@@ -0,0 +1,127 @@
+# FSM Compaction Design
+
+## Objective
+
+Run MVCC compaction automatically for the FSM store without breaking:
+
+1. Ongoing transactional reads.
+2. Read-modify-write flows that depend on a stable read timestamp.
+3. Raft apply progress during backlog replay or elections.
+
+## Problem
+
+The current FSM store used by `main.go` and `cmd/server/demo.go` is the
+in-memory `mvccStore`.
+
+1. It keeps all MVCC versions in memory.
+2. `Compact(ctx, minTS)` exists, but nothing calls it in production startup.
+3. The API still supports explicit historical reads (`GetAt`, `ScanAt`), so
+   naive compaction can silently change query semantics.
+
+## Design
+
+### 1. Background Compactor
+
+Add a background `kv.FSMCompactor` that runs on a timer instead of the Raft
+apply path.
+
+Why:
+
+1. Compaction is potentially heavy.
+2. Raft apply must stay simple and deterministic.
+3. Compaction should be skipped while a node is still catching up.
+
+### 2. Runtime Eligibility Checks
+
+For each Raft runtime, compaction only runs when:
+
+1. `fsm_pending == 0`
+2. `applied_index >= commit_index`
+3. Raft state is not `Candidate`
+
+This avoids compacting while the local FSM is still behind or unstable.
+
+### 3. Retention Watermark
+
+The compactor computes a safe `minTS` from:
+
+1. A retention window based on HLC wall time.
+2. The oldest actively pinned read timestamp.
+
+Effective rule:
+
+1. Start with `now - retentionWindow` in HLC form.
+2. If an older in-flight read exists, clamp the watermark to just before that
+   read timestamp.
+
+### 4. Active Timestamp Tracking
+
+Longer read-modify-write flows pin their read timestamp through
+`kv.ActiveTimestampTracker`.
+
+The current implementation pins:
+
+1. Redis `MULTI` / `EXEC` transactions.
+2. Redis Lua script execution.
+3. Distribution `SplitRange`, which performs a catalog read-modify-write cycle.
+
+### 5. Historical Read Contract
+
+The in-memory store now tracks `minRetainedTS`.
+
+1. Reads older than `minRetainedTS` fail with `store.ErrReadTSCompacted`.
+2. This makes compaction explicit instead of returning silently incorrect
+   historical results.
+3. `minRetainedTS` is included in `mvccStore` snapshots so restart/restore keeps
+   the same retention boundary.
+
+### 6. Scope of the Current Implementation
+
+Implemented now:
+
+1. Background FSM compactor in `kv/compactor.go`
+2. Active timestamp tracker in `kv/active_timestamp_tracker.go`
+3. `minRetainedTS` enforcement for `mvccStore`
+4. Startup wiring in:
+   - `main.go`
+   - `cmd/server/demo.go`
+5. Redis and distribution server timestamp pinning
+
+Not implemented yet:
+
+1. Pebble FSM compaction logic
+2. Dedicated GC for transaction metadata keys such as commit/rollback records
+3. Prometheus metrics for compaction runs
+4. Full active timestamp pinning across every DynamoDB multi-step flow
+
+## Current Defaults
+
+The compactor currently uses:
+
+1. Interval: `30s`
+2. Retention window: `30m`
+3. Per-run timeout: `5s`
+
+These are code defaults and can be tuned later if we expose flags or config.
+
+## Tradeoffs
+
+### Pros
+
+1. MVCC history no longer grows forever in the default in-memory FSM path.
+2. Ongoing Redis transactions and scripted operations are protected.
+3. Historical reads fail explicitly once compacted away.
+
+### Cons
+
+1. Arbitrary old raw timestamp reads are no longer guaranteed forever.
+2. DynamoDB paths currently rely mostly on the retention window rather than
+   explicit timestamp pinning.
+3. Transaction metadata keys still need a dedicated GC pass.
+
+## Follow-Up Work
+
+1. Add compaction metrics and visibility in the monitoring registry.
+2. Add transaction-metadata GC for `!txn|cmt|...` and `!txn|rb|...` keys.
+3. Extend active timestamp pinning to the remaining long DynamoDB flows.
+4. Reuse the same contract when `PebbleStore` becomes the primary FSM store.