feat(db): add a table for each member's last applied RGW frontend

microceph/database/rgw_frontend.go

@@ -0,0 +1,67 @@
+package database
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+)
+
+// RgwFrontend holds a member's last successfully applied public frontend settings.
+type RgwFrontend struct {
+	Member  string
+	Port    int
+	SSLPort int
+	SSL     bool
+}
+
+// UpsertRGWFrontend records frontend settings alongside the member's service row.
+func UpsertRGWFrontend(ctx context.Context, tx *sql.Tx, member string, port, sslPort int, ssl bool) error {
+	_, err := tx.ExecContext(ctx, `
+INSERT INTO rgw_frontends (member_id, port, ssl_port, ssl)
+VALUES ((SELECT id FROM core_cluster_members WHERE name = ?), ?, ?, ?)
+ON CONFLICT(member_id) DO UPDATE SET
+  port = excluded.port, ssl_port = excluded.ssl_port, ssl = excluded.ssl
+WHERE port != excluded.port OR ssl_port != excluded.ssl_port OR ssl != excluded.ssl`, member, port, sslPort, ssl)
+	if err != nil {
+		return fmt.Errorf("failed to upsert rgw frontend for %s: %w", member, err)
+	}
+	return nil
+}
+
+// GetRGWFrontends returns the recorded frontend settings for cluster members.
+func GetRGWFrontends(ctx context.Context, tx *sql.Tx) ([]RgwFrontend, error) {
+	rows, err := tx.QueryContext(ctx, `
+SELECT m.name, f.port, f.ssl_port, f.ssl
+  FROM rgw_frontends f
+  JOIN core_cluster_members m ON m.id = f.member_id
+ ORDER BY m.name`)
+	if err != nil {
+		return nil, fmt.Errorf("failed to query rgw frontends: %w", err)
+	}
+	defer func() { _ = rows.Close() }()
+
+	var frontends []RgwFrontend
+	for rows.Next() {
+		var f RgwFrontend
+		err := rows.Scan(&f.Member, &f.Port, &f.SSLPort, &f.SSL)
+		if err != nil {
+			return nil, fmt.Errorf("failed to scan rgw frontend: %w", err)
+		}
+		frontends = append(frontends, f)
+	}
+	err = rows.Err()
+	if err != nil {
+		return nil, fmt.Errorf("rgw frontend rows error: %w", err)
+	}
+	return frontends, nil
+}
+
+// DeleteRGWFrontendByMember removes a member's frontend record if present.
+func DeleteRGWFrontendByMember(ctx context.Context, tx *sql.Tx, member string) error {
+	_, err := tx.ExecContext(ctx, `
+DELETE FROM rgw_frontends WHERE member_id = (SELECT id FROM core_cluster_members WHERE name = ?)`, member)
+	if err != nil {
+		return fmt.Errorf("failed to delete rgw frontend for %s: %w", member, err)
+	}
+	return nil
+}

microceph/database/rgw_frontend_extras_test.go

@@ -0,0 +1,189 @@
+package database
+
+import (
+	"context"
+	"database/sql"
+	"testing"
+
+	_ "github.com/mattn/go-sqlite3"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// setupRGWFrontendDB creates an in-memory SQLite database (with foreign keys
+// enabled) containing a minimal core_cluster_members table (the FK target) and
+// the real schemaUpdate11 migration that creates rgw_frontends. Two named
+// members are seeded so name->id resolution and cascade-delete can be tested.
+func setupRGWFrontendDB(t *testing.T) *sql.DB {
+	t.Helper()
+	// _foreign_keys=on is per-connection in SQLite; the DSN parameter ensures
+	// every pooled connection honours ON DELETE CASCADE.
+	db, err := sql.Open("sqlite3", ":memory:?_foreign_keys=on")
+	require.NoError(t, err)
+	db.SetMaxOpenConns(1)
+	t.Cleanup(func() { _ = db.Close() })
+
+	// Minimal core_cluster_members matching the columns the rgw_frontends
+	// migration and the name->id subqueries reference.
+	_, err = db.Exec(`
+CREATE TABLE core_cluster_members (
+  id   INTEGER PRIMARY KEY AUTOINCREMENT NOT NULL,
+  name TEXT NOT NULL UNIQUE
+);
+INSERT INTO core_cluster_members (name) VALUES ('node-a');
+INSERT INTO core_cluster_members (name) VALUES ('node-b');
+`)
+	require.NoError(t, err)
+
+	tx, err := db.BeginTx(context.Background(), nil)
+	require.NoError(t, err)
+	err = schemaUpdate11(context.Background(), tx)
+	require.NoError(t, err)
+	require.NoError(t, tx.Commit())
+
+	return db
+}
+
+// TestUpsertRGWFrontendAndGetAll verifies an upsert is persisted and read back
+// keyed by member name, with ports and the TLS flag intact.
+func TestUpsertRGWFrontendAndGetAll(t *testing.T) {
+	db := setupRGWFrontendDB(t)
+	ctx := context.Background()
+
+	tx, err := db.BeginTx(ctx, nil)
+	require.NoError(t, err)
+	err = UpsertRGWFrontend(ctx, tx, "node-a", 80, 0, false)
+	require.NoError(t, err)
+	require.NoError(t, tx.Commit())
+
+	tx2, err := db.BeginTx(ctx, nil)
+	require.NoError(t, err)
+	defer func() { _ = tx2.Rollback() }()
+
+	frontends, err := GetRGWFrontends(ctx, tx2)
+	require.NoError(t, err)
+	require.Len(t, frontends, 1)
+	assert.Equal(t, "node-a", frontends[0].Member)
+	assert.Equal(t, 80, frontends[0].Port)
+	assert.Equal(t, 0, frontends[0].SSLPort)
+	assert.False(t, frontends[0].SSL)
+}
+
+// TestUpsertRGWFrontendOverwrites verifies a second upsert for the same member
+// replaces the row (cert rotation / port change), not a duplicate.
+func TestUpsertRGWFrontendOverwrites(t *testing.T) {
+	db := setupRGWFrontendDB(t)
+	ctx := context.Background()
+
+	lockTx(t, db, func(tx *sql.Tx) {
+		require.NoError(t, UpsertRGWFrontend(ctx, tx, "node-a", 80, 0, false))
+	})
+	lockTx(t, db, func(tx *sql.Tx) {
+		require.NoError(t, UpsertRGWFrontend(ctx, tx, "node-a", 8080, 443, true))
+	})
+
+	tx, err := db.BeginTx(ctx, nil)
+	require.NoError(t, err)
+	defer func() { _ = tx.Rollback() }()
+
+	frontends, err := GetRGWFrontends(ctx, tx)
+	require.NoError(t, err)
+	require.Len(t, frontends, 1, "upsert must replace, not append")
+	assert.Equal(t, 8080, frontends[0].Port)
+	assert.Equal(t, 443, frontends[0].SSLPort)
+	assert.True(t, frontends[0].SSL)
+}
+
+// TestGetRGWFrontendsMultipleMembers verifies the read returns each member's
+// frontend keyed by name.
+func TestGetRGWFrontendsMultipleMembers(t *testing.T) {
+	db := setupRGWFrontendDB(t)
+	ctx := context.Background()
+
+	lockTx(t, db, func(tx *sql.Tx) {
+		require.NoError(t, UpsertRGWFrontend(ctx, tx, "node-a", 80, 0, false))
+		require.NoError(t, UpsertRGWFrontend(ctx, tx, "node-b", 8443, 443, true))
+	})
+
+	tx, err := db.BeginTx(ctx, nil)
+	require.NoError(t, err)
+	defer func() { _ = tx.Rollback() }()
+
+	frontends, err := GetRGWFrontends(ctx, tx)
+	require.NoError(t, err)
+	require.Len(t, frontends, 2)
+
+	byName := map[string]RgwFrontend{}
+	for _, f := range frontends {
+		byName[f.Member] = f
+	}
+	assert.Contains(t, byName, "node-a")
+	assert.Contains(t, byName, "node-b")
+	assert.True(t, byName["node-b"].SSL)
+}
+
+// TestDeleteRGWFrontendByMember verifies the row is removed and that deleting a
+// missing row is not an error (idempotent scale-down / retry).
+func TestDeleteRGWFrontendByMember(t *testing.T) {
+	db := setupRGWFrontendDB(t)
+	ctx := context.Background()
+
+	lockTx(t, db, func(tx *sql.Tx) {
+		require.NoError(t, UpsertRGWFrontend(ctx, tx, "node-a", 80, 0, false))
+	})
+
+	lockTx(t, db, func(tx *sql.Tx) {
+		require.NoError(t, DeleteRGWFrontendByMember(ctx, tx, "node-a"))
+	})
+
+	// Read in its own (rolled-back) transaction, released before the re-delete
+	// below so the pool reuses the same :memory: connection.
+	func() {
+		tx, err := db.BeginTx(ctx, nil)
+		require.NoError(t, err)
+		frontends, err := GetRGWFrontends(ctx, tx)
+		require.NoError(t, err)
+		assert.Empty(t, frontends)
+		require.NoError(t, tx.Rollback())
+	}()
+
+	// Deleting an already-absent row must succeed (idempotent).
+	lockTx(t, db, func(tx *sql.Tx) {
+		require.NoError(t, DeleteRGWFrontendByMember(ctx, tx, "node-a"))
+	})
+}
+
+// TestRGWFrontendCascadeOnMemberRemoval verifies the ON DELETE CASCADE foreign
+// key removes the frontend row when the cluster member is deleted.
+func TestRGWFrontendCascadeOnMemberRemoval(t *testing.T) {
+	db := setupRGWFrontendDB(t)
+	ctx := context.Background()
+
+	lockTx(t, db, func(tx *sql.Tx) {
+		require.NoError(t, UpsertRGWFrontend(ctx, tx, "node-a", 80, 0, false))
+	})
+
+	_, err := db.Exec(`DELETE FROM core_cluster_members WHERE name = 'node-a'`)
+	require.NoError(t, err)
+
+	tx, err := db.BeginTx(ctx, nil)
+	require.NoError(t, err)
+	defer func() { _ = tx.Rollback() }()
+
+	var count int
+	err = tx.QueryRowContext(ctx, `SELECT COUNT(*) FROM rgw_frontends`).Scan(&count)
+	require.NoError(t, err)
+	assert.Zero(t, count, "removing the member must cascade-delete its frontend row")
+}
+
+func TestRGWFrontendRejectsUnknownMember(t *testing.T) {
+	db := setupRGWFrontendDB(t)
+	ctx := context.Background()
+	lockTx(t, db, func(tx *sql.Tx) {
+		err := UpsertRGWFrontend(ctx, tx, "unknown", 80, 0, false)
+		require.Error(t, err)
+		frontends, err := GetRGWFrontends(ctx, tx)
+		require.NoError(t, err)
+		assert.Empty(t, frontends)
+	})
+}

microceph/database/schema.go

@@ -23,6 +23,7 @@ var SchemaExtensions = []cluster.Update{
 	schemaUpdate8,
 	schemaUpdate9,
 	schemaUpdate10,
+	schemaUpdate11,
 }
 
 // getClusterTableName returns the name of the table that holds the record of cluster members from sqlite_master.
@@ -331,3 +332,22 @@ END;
 
 	return err
 }
+
+// schemaUpdate11 records each member's last successfully applied RGW frontend.
+// TLS material stays on the member and is never replicated in this table.
+func schemaUpdate11(ctx context.Context, tx *sql.Tx) error {
+	stmt := `
+CREATE TABLE rgw_frontends (
+  id         INTEGER PRIMARY KEY AUTOINCREMENT NOT NULL,
+  member_id  INTEGER NOT NULL,
+  port       INTEGER NOT NULL DEFAULT 0,
+  ssl_port   INTEGER NOT NULL DEFAULT 0,
+  ssl        INTEGER NOT NULL DEFAULT 0,
+  FOREIGN KEY (member_id) REFERENCES "core_cluster_members" (id) ON DELETE CASCADE,
+  UNIQUE(member_id)
+);
+  `
+	_, err := tx.ExecContext(ctx, stmt)
+
+	return err
+}