Add 'promote' flag to config (#306)

Author: benbjohnson

cmd/litefs/config.go

@@ -104,6 +104,10 @@ type LeaseConfig struct {
 	// Replicas in a state lease should set this to false.
 	Candidate bool `yaml:"candidate"`
 
+	// If true & node is a candidate, it will attempt to promote itself
+	// automatically once it connects to the cluster and syncs.
+	Promote bool `yaml:"promote"`
+
 	// After disconnect, time before node tries to reconnect to primary or
 	// becomes primary itself.
 	ReconnectDelay time.Duration `yaml:"reconnect-delay"`

cmd/litefs/mount_linux.go

@@ -13,6 +13,7 @@ import (
 	"regexp"
 	"strings"
 	"sync"
+	"time"
 
 	"github.com/mattn/go-shellwords"
 	"github.com/superfly/litefs"
@@ -237,18 +238,32 @@ func (c *MountCommand) Run(ctx context.Context) (err error) {
 		case <-c.Store.ReadyCh():
 			log.Printf("connected to cluster, ready")
 		}
-	}
 
-	// Execute subcommand, if specified in config.
-	if err := c.execCmd(ctx); err != nil {
-		return fmt.Errorf("cannot exec: %w", err)
+		// Automatically promote the server if requested & it is a candidate.
+		if c.Config.Lease.Promote {
+			if c.Store.Candidate() {
+				log.Printf("node is a candidate, automatically promoting to primary")
+				if err := c.promote(ctx); err != nil {
+					return fmt.Errorf("promote: %w", err)
+				}
+			} else {
+				log.Printf("node is not a candidate, skipping automatic promotion")
+			}
+		}
 	}
 
+	// Start the proxy server before the subcommand in case we need to hold
+	// requests after we promote but before the server is ready.
 	if c.ProxyServer != nil {
 		c.ProxyServer.Serve()
 		log.Printf("proxy server listening on: %s", c.ProxyServer.URL())
 	}
 
+	// Execute subcommand, if specified in config.
+	if err := c.execCmd(ctx); err != nil {
+		return fmt.Errorf("cannot exec: %w", err)
+	}
+
 	return nil
 }
 
@@ -395,4 +410,35 @@ func (c *MountCommand) execCmd(ctx context.Context) error {
 	return nil
 }
 
+// promote issues a lease handoff request to the current primary.
+func (c *MountCommand) promote(ctx context.Context) (err error) {
+	isPrimary, info := c.Store.PrimaryInfo()
+	if isPrimary {
+		log.Printf("node is already primary, skipping promotion")
+		return nil
+	}
+
+	client := http.NewClient()
+	if err := client.Handoff(ctx, info.AdvertiseURL, c.Store.ID()); err != nil {
+		return fmt.Errorf("handoff: %w", err)
+	}
+
+	// Wait for the local node to become primary.
+	ticker := time.NewTicker(1 * time.Millisecond)
+	defer ticker.Stop()
+	timeout := time.NewTicker(10 * time.Second)
+	defer timeout.Stop()
+
+	for {
+		select {
+		case <-timeout.C:
+			return fmt.Errorf("timed out waiting for promotion")
+		case <-ticker.C:
+			if c.Store.IsPrimary() {
+				return nil
+			}
+		}
+	}
+}
+
 var expvarOnce sync.Once

cmd/litefs/mount_test.go

@@ -1299,6 +1299,46 @@ func TestMultiNode_Handoff(t *testing.T) {
 	}
 }
 
+func TestMultiNode_Autopromotion(t *testing.T) {
+	cmd0 := newMountCommand(t, t.TempDir(), nil)
+	runMountCommand(t, cmd0)
+	waitForPrimary(t, cmd0)
+
+	// Create a simple table with a single value.
+	db0 := testingutil.OpenSQLDB(t, filepath.Join(cmd0.Config.FUSE.Dir, "db"))
+	if _, err := db0.Exec(`CREATE TABLE t (x)`); err != nil {
+		t.Fatal(err)
+	} else if _, err := db0.Exec(`INSERT INTO t VALUES (100)`); err != nil {
+		t.Fatal(err)
+	}
+
+	// Start a new instance with autopromotion enabled.
+	cmd1 := newMountCommand(t, t.TempDir(), cmd0)
+	cmd1.Config.Lease.Promote = true
+	runMountCommand(t, cmd1)
+	db1 := testingutil.OpenSQLDB(t, filepath.Join(cmd1.Config.FUSE.Dir, "db"))
+
+	// Ensure we can update the new primary node.
+	if _, err := db1.Exec(`INSERT INTO t VALUES (200)`); err != nil {
+		t.Fatal(err)
+	}
+
+	// Ensure the data exists on both nodes.
+	waitForSync(t, "db", cmd0, cmd1)
+	var sum int
+	if err := db1.QueryRow(`SELECT SUM(x) FROM t`).Scan(&sum); err != nil {
+		t.Fatal(err)
+	} else if got, want := sum, 300; got != want {
+		t.Fatalf("sum=%d, want %d", got, want)
+	}
+
+	if err := db0.QueryRow(`SELECT SUM(x) FROM t`).Scan(&sum); err != nil {
+		t.Fatal(err)
+	} else if got, want := sum, 300; got != want {
+		t.Fatalf("sum=%d, want %d", got, want)
+	}
+}
+
 func TestMultiNode_StaticLeaser(t *testing.T) {
 	dir0, dir1 := t.TempDir(), t.TempDir()
 	cmd0 := newMountCommand(t, dir0, nil)