feat(replication): epoch fencing and manual promotion (ADR 0002)

api/proto/beastdb.proto

@@ -17,6 +17,8 @@ service BeastDBService {
 service ReplicationService {
   rpc StreamWAL (StreamWALRequest) returns (stream WALRecordMessage);
   rpc Ack (ReplicationAck) returns (AckResponse);
+  // Fence freezes a leader's writes at a new epoch and reports its final LSN.
+  rpc Fence (FenceRequest) returns (FenceResponse);
 }
 
 message GetRequest {
@@ -66,22 +68,45 @@ message KeyValueResponse {
 message StreamWALRequest {
   string replica_id = 1;
   uint64 from_lsn = 2;
+  // Highest leader epoch the follower has seen. A leader that sees a higher
+  // epoch than its own steps down instead of serving.
+  uint64 epoch = 3;
 }
 
 message WALRecordMessage {
   uint64 lsn = 1;
   uint32 op_type = 2;
   bytes key = 3;
   bytes value = 4;
+  // Epoch of the leader that produced this record; followers reject lower
+  // epochs than their own.
+  uint64 epoch = 5;
 }
 
 message ReplicationAck {
   string replica_id = 1;
   uint64 acknowledged_lsn = 2;
+  uint64 epoch = 3;
 }
 
 message AckResponse {
   bool success = 1;
+  // The leader's epoch and current LSN, so an idle follower can tell whether it is caught up.
+  uint64 epoch = 2;
+  uint64 leader_lsn = 3;
+}
+
+message FenceRequest {
+  // New epoch the candidate will lead at; must exceed the leader's epoch.
+  uint64 epoch = 1;
+  // Replica asking to be promoted. Only one candidate can fence a given epoch.
+  string candidate_id = 2;
+}
+
+message FenceResponse {
+  // Final LSN of the fenced leader: it accepts no writes beyond this.
+  uint64 final_lsn = 1;
+  uint64 epoch = 2;
 }
 
 enum BatchOpType {

cmd/beastctl/backup_server_compat_test.go

@@ -0,0 +1,53 @@
+package main
+
+import (
+	"fmt"
+	"path/filepath"
+	"testing"
+
+	"github.com/ChromaBeast/beastdb/internal/api"
+	"github.com/ChromaBeast/beastdb/internal/backup"
+)
+
+// A backup made by the running server (internal/backup) must be accepted by the
+// offline tools: restore is still a beastctl step.
+func TestServerMadeBackupVerifiesAndRestoresWithBeastctl(t *testing.T) {
+	dir := t.TempDir()
+	eng, err := api.NewEngine(filepath.Join(dir, "beast.bin"), filepath.Join(dir, "beast.wal"), 32)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer eng.Close()
+	const n = 150
+	for i := uint64(1); i <= n; i++ {
+		if err := eng.Put(i, []byte(fmt.Sprintf("payload-%d", i))); err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	info, err := backup.Create(eng, filepath.Join(dir, "backups"), backup.KindManual, "test")
+	if err != nil {
+		t.Fatal(err)
+	}
+	bundle := filepath.Join(dir, "backups", info.ID)
+
+	if err := runVerify(bundle); err != nil {
+		t.Fatalf("beastctl verify rejected a server-made backup: %v", err)
+	}
+	res, err := runRestore(bundle, filepath.Join(dir, "restored"), false)
+	if err != nil {
+		t.Fatalf("beastctl restore failed on a server-made backup: %v", err)
+	}
+	restored, err := api.NewEngine(filepath.Join(dir, "restored", "beast.bin"), filepath.Join(dir, "restored", "beast.wal"), 32)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer restored.Close()
+	for i := uint64(1); i <= n; i++ {
+		v, ok, err := restored.Get(i)
+		if err != nil || !ok || string(v) != fmt.Sprintf("payload-%d", i) {
+			t.Fatalf("record %d after restore: %q %v %v", i, v, ok, err)
+		}
+	}
+	_ = res
+}

cmd/server/main.go

@@ -14,8 +14,10 @@ import (
 	"time"
 
 	"github.com/ChromaBeast/beastdb/internal/api"
+	"github.com/ChromaBeast/beastdb/internal/backup"
 	"github.com/ChromaBeast/beastdb/internal/logbuf"
 	"github.com/ChromaBeast/beastdb/internal/replication"
+	"github.com/ChromaBeast/beastdb/internal/runtimecfg"
 	"github.com/ChromaBeast/beastdb/internal/web"
 	"github.com/ChromaBeast/beastdb/internal/web/handler"
 	"google.golang.org/grpc"
@@ -72,15 +74,26 @@ func main() {
 	if err != nil {
 		log.Fatalf("Failed to initialize database engine: %v", err)
 	}
-	if *role != "leader" {
-		engine.SetReadOnly(true)
-		log.Printf("🔒 Node running in follower mode: engine write path set to read-only.")
-	}
 
 	ckptCtx, stopCkpt := context.WithCancel(context.Background())
 	defer stopCkpt()
 	engine.StartAutoCheckpoint(ckptCtx, 10*time.Second, int64(*walCheckpointMB)<<20)
 
+	// Settings an admin can change in Studio without a restart; saved values
+	// in runtime-settings.json override the matching flags.
+	runtimeCfg, rtErr := runtimecfg.Load(*dataDir, runtimecfg.Values{WALCheckpointMB: *walCheckpointMB, LoginMaxAttempts: 5, LoginWindowSeconds: 30, BackupKeep: 7})
+	if rtErr != nil {
+		log.Printf("Warning: ignoring runtime settings file: %v", rtErr)
+	}
+	runtimeCfg.Subscribe(func(v runtimecfg.Values) { engine.SetCheckpointThreshold(int64(v.WALCheckpointMB) << 20) })
+
+	// Server-managed backups (create, verify, download, optional schedule).
+	backups := backup.NewManager(engine, filepath.Join(*dataDir, "backups"))
+	runtimeCfg.Subscribe(func(v runtimecfg.Values) {
+		backups.Configure(time.Duration(v.BackupIntervalMinutes)*time.Minute, v.BackupKeep)
+	})
+	backups.Start(ckptCtx)
+
 	tokenStore := initTokenStore(engine)
 	grpcOpts := api.MessageSizeServerOptions()
 	// Metrics interceptors run first so they also observe auth rejections.
@@ -105,12 +118,25 @@ func main() {
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 
+	// Every node serves the replication service; it hands out WAL only while the
+	// node is a leader (or a fenced former leader). The persisted epoch and role
+	// in <data-dir>/cluster-state.json decide the starting role, so a restart
+	// cannot undo a promotion or un-fence an old leader.
+	flagRole := replication.RoleFollower
 	if *role == "leader" {
-		repServer := replication.NewLeaderServer(engine.WALPath())
-		repServer.RegisterService(grpcServer.RawServer())
-		engine.SetCommitObserver(repServer)
-		replicaSource = repServer
-	} else {
+		flagRole = replication.RoleLeader
+	}
+	repServer := replication.NewLeaderServer(engine.WALPath())
+	repServer.RegisterService(grpcServer.RawServer())
+	cluster, err := replication.NewCluster(engine, *dataDir, *replicaID, flagRole, repServer)
+	if err != nil {
+		log.Fatalf("FATAL: %v", err)
+	}
+	replicaSource = repServer
+	clusterControl = cluster
+	switch cluster.Role() {
+	case replication.RoleFollower:
+		log.Printf("🔒 Node running in follower mode at epoch %d: engine write path set to read-only.", cluster.Epoch())
 		clientCreds, err := loadClientTLS(*tlsCert, *tlsKey, *tlsCA)
 		if err != nil {
 			log.Fatalf("Failed to initialize follower client TLS: %v", err)
@@ -120,19 +146,9 @@ func main() {
 			log.Fatalf("Failed to dial leader node: %v", err)
 		}
 		defer conn.Close()
-		followerClient := replication.NewFollowerClient(*replicaID, engine, conn)
-		go func() {
-			for ctx.Err() == nil {
-				if err := followerClient.SyncLoop(ctx); err != nil && ctx.Err() == nil {
-					log.Printf("Follower sync stream dropped: %v. Reconnecting in 1s...", err)
-					select {
-					case <-time.After(time.Second):
-					case <-ctx.Done():
-						return
-					}
-				}
-			}
-		}()
+		cluster.RunFollower(ctx, replication.NewFollowerClient(*replicaID, engine, conn))
+	case replication.RoleFenced:
+		log.Printf("🔒 Node was fenced at epoch %d: it stays read-only. Wipe its data directory to rejoin it as a follower.", cluster.Epoch())
 	}
 
 	var partitions []handler.PartitionEntry
@@ -159,7 +175,9 @@ func main() {
 
 	logs.AddSecret(nonDefault(*adminPassword)) // dev mode generates its password after flag parsing
 	webOpts := web.Options{
-		Logs: logs,
+		Logs:    logs,
+		Runtime: runtimeCfg,
+		Backups: backups,
 		Settings: handler.RuntimeSettings{
 			GRPCAddr: grpcBind, WebAddr: *webAddr, DataDir: *dataDir, PoolSize: *poolSize,
 			WALCheckpointMB: *walCheckpointMB, ReplicaID: *replicaID, LeaderAddr: *leaderAddr,
@@ -168,6 +186,7 @@ func main() {
 			PartitionConfigPath: cfgPath, DevMode: *devMode, InsecureAuthOverride: *insecureAuth,
 		},
 	}
+	webOpts.Cluster = clusterControl
 	stopWeb := startWebServer(*webAddr, *dataDir, *role, *adminPassword, engine, partitions, tokenStore, webOpts, *apiToken, os.Getenv("BEASTDB_METRICS_TOKEN"))
 	defer stopWeb()
 

cmd/server/web.go

@@ -4,6 +4,7 @@ import (
 	"log"
 
 	"github.com/ChromaBeast/beastdb/internal/api"
+	"github.com/ChromaBeast/beastdb/internal/audit"
 	"github.com/ChromaBeast/beastdb/internal/web"
 	"github.com/ChromaBeast/beastdb/internal/web/auth"
 	"github.com/ChromaBeast/beastdb/internal/web/handler"
@@ -14,6 +15,9 @@ import (
 // It stays nil on followers (they do not see other replicas).
 var replicaSource handler.ReplicationSource
 
+// clusterControl is set by main; it carries the node's epoch, role and promotion.
+var clusterControl handler.ClusterControl
+
 func startWebServer(webAddr, dataDir, role, adminPassword string, engine *api.Engine, partitions []handler.PartitionEntry, tokenStore *auth.TokenStore, extra web.Options, metricsTokens ...string) func() {
 	if webAddr == "" {
 		return func() {}
@@ -24,6 +28,12 @@ func startWebServer(webAddr, dataDir, role, adminPassword string, engine *api.En
 		log.Fatalf("Failed to initialize session secret: %v", secretErr)
 	}
 
+	auditLog, auditErr := audit.Open(dataDir)
+	if auditErr != nil {
+		log.Fatalf("Failed to open audit log: %v", auditErr)
+	}
+	extra.Audit = auditLog
+
 	webSrv, webErr := web.NewServerWithOptions(webAddr, engine, secret, role, Version, partitions, withBase(withReplication(extra), dataDir, metricsTokens), tokenStore)
 	if webErr != nil {
 		log.Fatalf("Failed to create web server: %v", webErr)
@@ -47,6 +57,7 @@ func startWebServer(webAddr, dataDir, role, adminPassword string, engine *api.En
 	go webSrv.Serve()
 	return func() {
 		webSrv.Shutdown()
+		_ = auditLog.Close()
 	}
 }