Narrow AUTO_INCREMENT ids from cluster range claims; log-cursor anti-entropy; per-row last-writer-wins (fixes #170)

admin/autoinc.go

@@ -0,0 +1,110 @@
+package admin
+
+import (
+	"net/http"
+
+	"github.com/maxpert/marmot/db"
+	marmotgrpc "github.com/maxpert/marmot/grpc"
+)
+
+// handleAutoIncVotes handles GET /admin/cluster/autoinc/votes: whether this
+// node's AUTO_INCREMENT claim votes are held (db.AutoIncHoldTable). While
+// they are, the node declines every claim at PREPARE, so narrow
+// AUTO_INCREMENT inserts that need a new range can return 1205 until enough
+// members answered its merge. It is the "unheld" check of the
+// membership-change procedure (handleAutoIncSync), and a readiness signal
+// after a node's first start.
+func (h *AdminHandlers) handleAutoIncVotes(w http.ResponseWriter, r *http.Request) {
+	held, err := db.NewAutoIncClaimStore(h.dbManager.GetSystemDatabase()).VotesHeld()
+	if err != nil {
+		writeErrorResponse(w, http.StatusInternalServerError, err.Error())
+		return
+	}
+	writeJSONResponse(w, map[string]interface{}{"votes_held": held}, false, "")
+}
+
+// handleAutoIncReleaseVotes handles POST
+// /admin/cluster/autoinc/release-votes?accept_risk=duplicate-ids.
+//
+// It releases this node's held AUTO_INCREMENT claim votes after merging the
+// bases of every peer that answers, without the safety condition the
+// automatic merge waits for (grpc.ForceReleaseAutoIncVotes). It is for a
+// cluster held beyond its failure model, and it can make the cluster issue
+// the same id twice if a member holding a newer claim does not answer, so the
+// caller must name that risk in accept_risk.
+func (h *AdminHandlers) handleAutoIncReleaseVotes(w http.ResponseWriter, r *http.Request) {
+	if r.URL.Query().Get("accept_risk") != marmotgrpc.ForceReleaseRiskToken {
+		writeErrorResponse(w, http.StatusBadRequest,
+			"releasing held AUTO_INCREMENT claim votes without a safe merge can issue the same id twice "+
+				"if a member holding a newer claim does not answer; pass accept_risk="+marmotgrpc.ForceReleaseRiskToken)
+		return
+	}
+	registry := h.resolveRegistryOrError(w)
+	if registry == nil {
+		return
+	}
+	client := h.server.GetGossipProtocol().GetClient()
+	if client == nil {
+		writeErrorResponse(w, http.StatusServiceUnavailable, "cluster client not started")
+		return
+	}
+
+	answered, wasHeld, err := marmotgrpc.ForceReleaseAutoIncVotes(r.Context(), registry.GetLocalNodeID(), h.dbManager, registry, client)
+	if err != nil {
+		writeErrorResponse(w, http.StatusInternalServerError, err.Error())
+		return
+	}
+	writeJSONResponse(w, map[string]interface{}{
+		"was_held":       wasHeld,
+		"released":       wasHeld,
+		"peers_answered": answered,
+	}, false, "")
+}
+
+// handleAutoIncSync handles POST /admin/cluster/autoinc/sync.
+//
+// It runs the AUTO_INCREMENT claim base sync now on this node and on every
+// member of its membership (grpc.SyncAutoIncBasesEverywhere), and reports
+// per node whether it answered, whether its votes are held and whether it
+// reached every member. It is the confirmation step of the membership-change
+// procedure: add or remove one node, wait until it is ALIVE and unheld, run
+// this - re-run until the report is complete - and make the next change only
+// when "complete" is true.
+func (h *AdminHandlers) handleAutoIncSync(w http.ResponseWriter, r *http.Request) {
+	registry := h.resolveRegistryOrError(w)
+	if registry == nil {
+		return
+	}
+	client := h.server.GetGossipProtocol().GetClient()
+	if client == nil {
+		writeErrorResponse(w, http.StatusServiceUnavailable, "cluster client not started")
+		return
+	}
+	outcomes := marmotgrpc.SyncAutoIncBasesEverywhere(r.Context(), registry.GetLocalNodeID(), h.dbManager, registry, client)
+	writeJSONResponse(w, autoIncSyncBody(outcomes), false, "")
+}
+
+// autoIncSyncBody is the sync command's answer: one entry per member, and
+// complete when every member answered, is unheld and reached every member.
+func autoIncSyncBody(outcomes []marmotgrpc.AutoIncSyncOutcome) map[string]interface{} {
+	complete := len(outcomes) > 0
+	nodes := make([]map[string]interface{}, 0, len(outcomes))
+	for _, o := range outcomes {
+		node := map[string]interface{}{
+			"node_id":  o.NodeID,
+			"answered": o.Err == nil,
+			"complete": o.Complete(),
+		}
+		if o.Err != nil {
+			node["error"] = o.Err.Error()
+		} else {
+			node["votes_held"] = o.Report.VotesHeld
+			node["members"] = o.Report.Members
+			node["reached"] = o.Report.Reached
+			node["reached_every_member"] = o.Report.ReachedEveryMember()
+		}
+		complete = complete && o.Complete()
+		nodes = append(nodes, node)
+	}
+	return map[string]interface{}{"complete": complete, "nodes": nodes}
+}

admin/autoinc_test.go

@@ -0,0 +1,127 @@
+package admin
+
+import (
+	"errors"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+
+	"github.com/maxpert/marmot/cfg"
+	"github.com/maxpert/marmot/db"
+	marmotgrpc "github.com/maxpert/marmot/grpc"
+	"github.com/maxpert/marmot/hlc"
+)
+
+// TestAutoIncReleaseVotesRequiresTheRiskToken pins that the forced release is
+// never reached by accident: without accept_risk naming the risk, or with any
+// other value, it is refused before anything is touched, and the refusal
+// states the risk.
+//
+// Mutation: drop the accept_risk check. The handler reaches the nil server
+// and panics instead of answering 400.
+func TestAutoIncReleaseVotesRequiresTheRiskToken(t *testing.T) {
+	h := &AdminHandlers{}
+	for _, query := range []string{"", "?accept_risk=yes"} {
+		rec := httptest.NewRecorder()
+		h.handleAutoIncReleaseVotes(rec, httptest.NewRequest(http.MethodPost, "/cluster/autoinc/release-votes"+query, nil))
+		if rec.Code != http.StatusBadRequest {
+			t.Fatalf("%q: status %d, want 400", query, rec.Code)
+		}
+		if body := rec.Body.String(); !strings.Contains(body, "same id twice") || !strings.Contains(body, marmotgrpc.ForceReleaseRiskToken) {
+			t.Fatalf("%q: the refusal does not state the risk and the token: %s", query, body)
+		}
+	}
+}
+
+// TestAutoIncEndpointsRequireTheClusterSecret pins that the release and
+// sync endpoints are refused on a cluster with no secret, and served only
+// with the right one on a cluster that has one - unlike the other admin
+// endpoints, which AuthMiddleware serves to anyone when no secret is set.
+//
+// Mutation: route the autoinc endpoints through chiAuthMiddleware only. The
+// no-secret request reaches the handler and "served without a cluster
+// secret" fires.
+func TestAutoIncEndpointsRequireTheClusterSecret(t *testing.T) {
+	saved := cfg.Config.Cluster.ClusterSecret
+	t.Cleanup(func() { cfg.Config.Cluster.ClusterSecret = saved })
+
+	mux := http.NewServeMux()
+	RegisterRoutes(mux, &AdminHandlers{})
+	do := func(path, secret string) int {
+		req := httptest.NewRequest(http.MethodPost, "/admin"+path, nil)
+		if secret != "" {
+			req.Header.Set("X-Marmot-Secret", secret)
+		}
+		rec := httptest.NewRecorder()
+		mux.ServeHTTP(rec, req)
+		return rec.Code
+	}
+
+	for _, path := range []string{"/cluster/autoinc/sync", "/cluster/autoinc/release-votes?accept_risk=" + marmotgrpc.ForceReleaseRiskToken} {
+		cfg.Config.Cluster.ClusterSecret = ""
+		if code := do(path, ""); code != http.StatusForbidden {
+			t.Fatalf("%s: status %d, want 403: served without a cluster secret", path, code)
+		}
+		cfg.Config.Cluster.ClusterSecret = "s3cret"
+		if code := do(path, ""); code != http.StatusUnauthorized {
+			t.Fatalf("%s: status %d without the secret, want 401", path, code)
+		}
+		if code := do(path, "wrong"); code != http.StatusUnauthorized {
+			t.Fatalf("%s: status %d with a wrong secret, want 401", path, code)
+		}
+	}
+}
+
+// TestAutoIncSyncBodyIsCompleteOnlyWhenEveryMemberIs pins the sync command's
+// verdict: complete only when every member answered, is unheld and reached
+// every member.
+func TestAutoIncSyncBodyIsCompleteOnlyWhenEveryMemberIs(t *testing.T) {
+	ready := marmotgrpc.AutoIncSyncOutcome{NodeID: 1, Report: marmotgrpc.AutoIncSyncReport{
+		NodeID: 1, Members: []uint64{1, 2}, Reached: []uint64{1, 2}}}
+	if body := autoIncSyncBody([]marmotgrpc.AutoIncSyncOutcome{ready, ready}); body["complete"] != true {
+		t.Fatalf("every member ready: %v", body)
+	}
+	down := marmotgrpc.AutoIncSyncOutcome{NodeID: 2, Err: errors.New("unreachable")}
+	body := autoIncSyncBody([]marmotgrpc.AutoIncSyncOutcome{ready, down})
+	if body["complete"] != false {
+		t.Fatalf("an unreachable member reported complete: %v", body)
+	}
+	nodes := body["nodes"].([]map[string]interface{})
+	if nodes[1]["answered"] != false || nodes[1]["error"] != "unreachable" {
+		t.Fatalf("the unreachable member's entry: %v", nodes[1])
+	}
+	if autoIncSyncBody(nil)["complete"] != false {
+		t.Fatal("no outcome at all reported complete")
+	}
+}
+
+// TestAutoIncVotesReportsTheHold pins the read-only readiness signal: a node
+// whose system database was just created holds its claim votes, and reports
+// them released once the hold row is gone.
+func TestAutoIncVotesReportsTheHold(t *testing.T) {
+	dm, err := db.NewDatabaseManager(t.TempDir(), 1, hlc.NewClock(1))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer dm.Close()
+	h := &AdminHandlers{dbManager: dm}
+
+	votesHeld := func() string {
+		rec := httptest.NewRecorder()
+		h.handleAutoIncVotes(rec, httptest.NewRequest(http.MethodGet, "/cluster/autoinc/votes", nil))
+		if rec.Code != http.StatusOK {
+			t.Fatalf("status %d: %s", rec.Code, rec.Body.String())
+		}
+		return rec.Body.String()
+	}
+	if body := votesHeld(); !strings.Contains(body, `"votes_held":true`) {
+		t.Fatalf("a fresh node must report its votes held: %s", body)
+	}
+	if _, err := dm.GetSystemDatabase().GetWriteDB().Exec("DELETE FROM " + db.AutoIncHoldTable); err != nil {
+		t.Fatal(err)
+	}
+	if body := votesHeld(); !strings.Contains(body, `"votes_held":false`) {
+		t.Fatalf("a released node must report its votes unheld: %s", body)
+	}
+}

admin/cluster.go

@@ -82,7 +82,15 @@ func (h *AdminHandlers) handleClusterHealth(w http.ResponseWriter, r *http.Reque
 	}, false, "")
 }
 
-// handleClusterReplication handles GET /admin/cluster/replication
+// handleClusterReplication handles GET /admin/cluster/replication. It
+// reports the log-pull replication state per peer and per database,
+// instead of the removed MetaStore.ReplicationState table: for each peer,
+// its last-consumed position in this node's log for the database
+// (MetaStore.ConsumedPositions), and, separately, this node's own pull
+// cursor into that peer's log (MetaStore.GetPullCursor) - the two directions
+// of replication progress between this node and the peer. The per-database
+// caught-up/stuck status (h.antiEntropyStatus) is not per peer: it is this
+// node's own view of the whole database.
 func (h *AdminHandlers) handleClusterReplication(w http.ResponseWriter, r *http.Request) {
 	registry := h.resolveRegistryOrError(w)
 	if registry == nil {
@@ -107,33 +115,26 @@ func (h *AdminHandlers) handleClusterReplication(w http.ResponseWriter, r *http.
 			if err != nil {
 				continue
 			}
-
 			metaStore := mdb.GetMetaStore()
-			state, _ := metaStore.GetReplicationState(member.NodeID, dbName)
-			localMaxTxn, _ := metaStore.GetMaxCommittedTxnID()
-
-			var lastAppliedTxnID uint64
-			var lastSyncTime string
-			var syncStatus string
-			if state != nil {
-				lastAppliedTxnID = state.LastAppliedTxnID
-				lastSyncTime = formatTimestamp(state.LastSyncTime)
-				syncStatus = state.SyncStatus.String()
-			} else {
-				syncStatus = "UNKNOWN"
-			}
 
-			lag := uint64(0)
-			if localMaxTxn > lastAppliedTxnID {
-				lag = localMaxTxn - lastAppliedTxnID
+			consumed, err := metaStore.ConsumedPositions()
+			if err != nil {
+				continue
+			}
+			cursor, err := metaStore.GetPullCursor(member.NodeID)
+			if err != nil {
+				continue
 			}
 
+			consumedPos, hasConsumed := consumed[member.NodeID]
+
 			dbStates = append(dbStates, map[string]interface{}{
-				"database":            dbName,
-				"last_applied_txn_id": lastAppliedTxnID,
-				"last_sync_time":      lastSyncTime,
-				"sync_status":         syncStatus,
-				"lag_txns":            lag,
+				"database":                 dbName,
+				"peer_consumed_seq":        consumedPos.Seq,
+				"peer_consumed_txn_id":     consumedPos.TxnID,
+				"peer_has_consumed_at_all": hasConsumed,
+				"local_pull_cursor_seq":    cursor.Seq,
+				"local_pull_cursor_txn_id": cursor.TxnID,
 			})
 		}
 
@@ -146,12 +147,32 @@ func (h *AdminHandlers) handleClusterReplication(w http.ResponseWriter, r *http.
 	}
 
 	writeJSONResponse(w, map[string]interface{}{
-		"peers":                 result,
-		"cluster_min_watermark": registry.GetClusterMinWatermark(),
-		"local_watermark":       registry.GetLocalWatermark(),
+		"peers":        result,
+		"anti_entropy": h.antiEntropyStatus(databases),
 	}, false, "")
 }
 
+// antiEntropyStatus reports, per database, anti-entropy's status as of its
+// last round: whether every current member's
+// log was reachable and fully pulled (CaughtUp), and how many transactions
+// are STUCK on a deterministic replay failure (LogPuller.StuckTxns). Empty
+// until anti-entropy is wired in (SetAntiEntropy).
+func (h *AdminHandlers) antiEntropyStatus(databases []string) []map[string]interface{} {
+	result := make([]map[string]interface{}, 0, len(databases))
+	if h.antiEntropy == nil {
+		return result
+	}
+	for _, dbName := range databases {
+		result = append(result, map[string]interface{}{
+			"database":        dbName,
+			"caught_up":       h.antiEntropy.CaughtUp(dbName),
+			"promotion_ready": h.antiEntropy.PromotionReady(dbName),
+			"stuck_txns":      h.antiEntropy.StuckTxnCount(dbName),
+		})
+	}
+	return result
+}
+
 // handleClusterRemove handles POST /admin/cluster/remove/{node_id}
 func (h *AdminHandlers) handleClusterRemove(w http.ResponseWriter, r *http.Request) {
 	nodeID, err := parsePeerNodeID(chi.URLParam(r, "nodeID"))

admin/handlers.go

@@ -19,8 +19,9 @@ var uiHTML embed.FS
 
 // AdminHandlers handles admin API endpoints for MetaStore operations
 type AdminHandlers struct {
-	server    *grpc.Server
-	dbManager *db.DatabaseManager
+	server      *grpc.Server
+	dbManager   *db.DatabaseManager
+	antiEntropy *grpc.AntiEntropyService // set once anti-entropy is constructed (SetAntiEntropy); nil until then
 }
 
 // NewAdminHandlers creates a new AdminHandlers instance
@@ -31,6 +32,16 @@ func NewAdminHandlers(server *grpc.Server, dbManager *db.DatabaseManager) *Admin
 	}
 }
 
+// SetAntiEntropy wires the running AntiEntropyService into the admin
+// surface, so cluster/replication status can report per-database caught-up
+// state and stuck transactions. marmot.go calls
+// this once anti-entropy is constructed, which happens after
+// NewAdminHandlers (anti-entropy needs the DatabaseManager and replication
+// handler that are wired after admin routes are registered).
+func (h *AdminHandlers) SetAntiEntropy(ae *grpc.AntiEntropyService) {
+	h.antiEntropy = ae
+}
+
 // ServeUI serves the admin UI HTML file
 func (h *AdminHandlers) ServeUI(w http.ResponseWriter, r *http.Request) {
 	// Serve the embedded HTML file