internal/rest/resources/control: Use independent context for join faiure cleanup

example/test/main.sh

@@ -1047,6 +1047,139 @@ test_self_deletion() {
   shutdown_systems
 }
 
+test_failed_join() {
+  echo "Testing failed join"
+
+  new_systems 4 --heartbeat 2s
+
+  # Bootstrap initial cluster.
+  microctl --state-dir "${test_dir}/c1" init "c1" 127.0.0.1:9001 --bootstrap
+
+  token_c2=$(microctl --state-dir "${test_dir}/c1" tokens add "c2")
+  token_c3=$(microctl --state-dir "${test_dir}/c1" tokens add "c3")
+  microctl --state-dir "${test_dir}/c2" init "c2" 127.0.0.1:9002 --token "${token_c2}"
+  microctl --state-dir "${test_dir}/c3" init "c3" 127.0.0.1:9003 --token "${token_c3}"
+
+  # Wait for cluster to stabilize
+  while [[ -n "$(microctl --state-dir "${test_dir}/c1" cluster list -f yaml | yq '.[] | select(.role == "PENDING")')" ]]; do
+    sleep 2
+  done
+
+  # An invalid schema patch makes the join of c4 fail after dqlite has already added it to the cluster.
+  echo "  -> Testing join of c4 with a failing schema update"
+  token_c4=$(microctl --state-dir "${test_dir}/c1" tokens add "c4")
+  echo "INTENTIONALLY INVALID SQL STATEMENT;" > "${test_dir}/c4/patch.global.sql"
+  if microctl --state-dir "${test_dir}/c4" init "c4" 127.0.0.1:9004 --token "${token_c4}"; then
+    echo "ERROR: Member c4 should have failed to join"
+    exit 1
+  fi
+
+  # The remaining cluster members keep connecting to the failed joiner for a while,
+  # which must not crash its daemon after its dqlite app has been closed.
+  echo "  -> Testing c4 daemon survives the failed join"
+  sleep 5
+  if ! microctl --state-dir "${test_dir}/c4" waitready --timeout 30; then
+    echo "ERROR: Daemon of member c4 did not survive the failed join"
+    exit 1
+  fi
+
+  echo "  -> Testing c4 is no longer present on the remaining cluster"
+  if [ "$(microctl --state-dir "${test_dir}/c1" cluster list -f csv | wc -l)" != "3" ]; then
+    echo "ERROR: Member c4 is still present on the remaining cluster"
+    microctl --state-dir "${test_dir}/c1" cluster list
+    exit 1
+  fi
+
+  echo "  -> Testing c4 can join after the failure is resolved"
+  rm -f "${test_dir}/c4/patch.global.sql"
+  token_c4=$(microctl --state-dir "${test_dir}/c1" tokens add "c4")
+  if ! microctl --state-dir "${test_dir}/c4" init "c4" 127.0.0.1:9004 --token "${token_c4}"; then
+    echo "ERROR: Member c4 failed to join after the failure was resolved"
+    exit 1
+  fi
+
+  if [ "$(microctl --state-dir "${test_dir}/c1" cluster list -f csv | wc -l)" != "4" ]; then
+    echo "ERROR: Member c4 is missing from the cluster after joining"
+    microctl --state-dir "${test_dir}/c1" cluster list
+    exit 1
+  fi
+
+  shutdown_systems
+}
+
+test_failed_join_client_timeout() {
+  echo "Testing failed join after the client timed out"
+
+  new_systems 4 --heartbeat 2s
+
+  # Bootstrap initial cluster.
+  microctl --state-dir "${test_dir}/c1" init "c1" 127.0.0.1:9001 --bootstrap
+
+  token_c2=$(microctl --state-dir "${test_dir}/c1" tokens add "c2")
+  token_c3=$(microctl --state-dir "${test_dir}/c1" tokens add "c3")
+  microctl --state-dir "${test_dir}/c2" init "c2" 127.0.0.1:9002 --token "${token_c2}"
+  microctl --state-dir "${test_dir}/c3" init "c3" 127.0.0.1:9003 --token "${token_c3}"
+
+  # Wait for cluster to stabilize
+  while [[ -n "$(microctl --state-dir "${test_dir}/c1" cluster list -f yaml | yq '.[] | select(.role == "PENDING")')" ]]; do
+    sleep 2
+  done
+
+  # A slow and then invalid schema patch makes the join of c4 fail several seconds after
+  # dqlite has already added it to the cluster, and after the client below has given up.
+  echo "  -> Testing join of c4 with a client that times out before the join fails"
+  token_c4=$(microctl --state-dir "${test_dir}/c1" tokens add "c4")
+  cat << EOF > "${test_dir}/c4/patch.global.sql"
+CREATE TEMP TABLE slow AS WITH RECURSIVE c(x) AS (SELECT 1 UNION ALL SELECT x+1 FROM c WHERE x < 3000000) SELECT x FROM c;
+INTENTIONALLY INVALID SQL STATEMENT;
+EOF
+
+  if curl -sS --max-time 1 --unix-socket "${test_dir}/c4/control.socket" \
+    -X POST -H "Content-Type: application/json" \
+    -d "{\"bootstrap\":false,\"join_token\":\"${token_c4}\",\"address\":\"127.0.0.1:9004\",\"name\":\"c4\",\"config\":{}}" \
+    http://control.socket/core/control; then
+    echo "ERROR: Join request of c4 should have timed out"
+    exit 1
+  fi
+
+  # Make sure the join got far enough to change the cluster state before the client gave up.
+  if [ "$(microctl --state-dir "${test_dir}/c1" cluster list -f csv | wc -l)" != "4" ]; then
+    echo "ERROR: Member c4 was not added to the cluster before the join request timed out"
+    microctl --state-dir "${test_dir}/c1" cluster list
+    exit 1
+  fi
+
+  # The join failure cleanup must still run, even though the join request context has expired.
+  echo "  -> Testing c4 is removed from the remaining cluster"
+  retries=0
+  while [ "$(microctl --state-dir "${test_dir}/c1" cluster list -f csv | wc -l)" != "3" ]; do
+    retries=$((retries + 1))
+    if [ "${retries}" -gt 30 ]; then
+      echo "ERROR: Member c4 is still present on the remaining cluster"
+      microctl --state-dir "${test_dir}/c1" cluster list
+      exit 1
+    fi
+
+    sleep 2
+  done
+
+  if ! microctl --state-dir "${test_dir}/c1" tokens add "c5" > /dev/null; then
+    echo "ERROR: Cluster membership is inconsistent after the failed join"
+    exit 1
+  fi
+
+  echo "  -> Testing c4 can join after the failure is resolved"
+  microctl --state-dir "${test_dir}/c4" waitready --timeout 30
+  rm -f "${test_dir}/c4/patch.global.sql"
+  token_c4=$(microctl --state-dir "${test_dir}/c1" tokens add "c4")
+  if ! microctl --state-dir "${test_dir}/c4" init "c4" 127.0.0.1:9004 --token "${token_c4}"; then
+    echo "ERROR: Member c4 failed to join after the failure was resolved"
+    exit 1
+  fi
+
+  shutdown_systems
+}
+
 # allow for running a specific set of tests
 TEST_RESULT="success"
 if [ "${1:-"all"}" = "all" ] || [ "${1}" = "" ]; then
@@ -1061,6 +1194,8 @@ if [ "${1:-"all"}" = "all" ] || [ "${1}" = "" ]; then
   run_test truststore_force_removal
   run_test parallel_joins
   run_test self_deletion
+  run_test failed_join
+  run_test failed_join_client_timeout
 elif [ "${1}" = "recover" ]; then
   run_test recover
 elif [ "${1}" = "tokens" ]; then
@@ -1081,6 +1216,10 @@ elif [ "${1}" = "parallel-join" ]; then
   run_test parallel_joins
 elif [ "${1}" = "self-deletion" ]; then
   run_test self_deletion
+elif [ "${1}" = "failed-join" ]; then
+  run_test failed_join
+elif [ "${1}" = "failed-join-client-timeout" ]; then
+  run_test failed_join_client_timeout
 elif [ "${1}" = "daemon-config" ]; then
   run_test daemon_config_api
 else

internal/db/dqlite.go

@@ -68,7 +68,9 @@ const (
 // Accept passes conn to dqlite. During normal operation the send completes
 // immediately. During a Restart the channel has no consumer yet, so the call
 // blocks for up to 10s, keeping clients "on hold" until the new dqlite
-// instance is ready, making the restart invisible to them.
+// instance is ready, making the restart invisible to them. If there is no
+// dqlite instance at all (e.g. after a failed bootstrap or join), the
+// connection is closed after the timeout.
 // acceptMu is held so that Restart can atomically rotate the channel.
 func (db *DqliteDB) Accept(conn net.Conn) {
 	db.acceptMu.RLock()
@@ -83,6 +85,23 @@ func (db *DqliteDB) Accept(conn net.Conn) {
 	}
 }
 
+// rotateAcceptCh swaps in a fresh accept channel and closes the old one, so that the
+// go-dqlite reader ranging over it exits. This must be done before closing a dqlite app,
+// as the reader otherwise keeps accepting connections and panics when it fails to dial
+// the closed app's local socket.
+// Accept() holds acceptMu for every send, so once the write lock is held there
+// can be no remaining senders on the old channel.
+func (db *DqliteDB) rotateAcceptCh() {
+	db.acceptMu.Lock()
+	oldAcceptCh := db.acceptCh
+	db.acceptCh = make(chan net.Conn)
+	db.acceptMu.Unlock()
+
+	if oldAcceptCh != nil {
+		close(oldAcceptCh)
+	}
+}
+
 // NewDB creates an empty db struct with no dqlite connection.
 func NewDB(ctx context.Context, serverCert func() *shared.CertInfo, clusterCert func() *shared.CertInfo, memberName func() string, failureDomain func() uint64, os types.OS, heartbeatInterval time.Duration) (*DqliteDB, error) {
 	shutdownCtx, shutdownCancel := context.WithCancel(ctx)
@@ -179,6 +198,8 @@ func (db *DqliteDB) Bootstrap(extensions types.Extensions, addr *url.URL, cluste
 	defer reverter.Fail()
 	reverter.Add(func() {
 		if db.dqlite != nil {
+			db.rotateAcceptCh()
+
 			closeErr := db.dqlite.Close()
 			if closeErr != nil {
 				db.log().Error("Failed to close dqlite app", slog.String("address", db.listenAddr.String()), slog.String("error", closeErr.Error()))
@@ -232,6 +253,8 @@ func (db *DqliteDB) Join(extensions types.Extensions, addr *url.URL, joinAddress
 	defer reverter.Fail()
 	reverter.Add(func() {
 		if db.dqlite != nil {
+			db.rotateAcceptCh()
+
 			closeErr := db.dqlite.Close()
 			if closeErr != nil {
 				db.log().Error("Failed to close dqlite app", slog.String("address", db.listenAddr.String()), slog.String("error", closeErr.Error()))
@@ -522,18 +545,7 @@ func (db *DqliteDB) Restart(extensions types.Extensions, clusterMembers map[stri
 	db.status = types.DatabaseOffline
 	db.statusLock.Unlock()
 
-	// Swap in a fresh accept channel for the next dqlite instance and close the
-	// old one so any stale go-dqlite reader ranging over it can exit.
-	// Accept() holds acceptMu for every send, so once the write lock is held there
-	// can be no remaining senders on the old channel.
-	db.acceptMu.Lock()
-	oldAcceptCh := db.acceptCh
-	db.acceptCh = make(chan net.Conn)
-	db.acceptMu.Unlock()
-
-	if oldAcceptCh != nil {
-		close(oldAcceptCh)
-	}
+	db.rotateAcceptCh()
 
 	if db.db != nil {
 		_ = db.db.Close()

internal/rest/resources/cluster.go

@@ -53,6 +53,9 @@ var clusterMemberCmd = types.Endpoint{
 var clusterMemberInternalCmd = types.Endpoint{
 	Path: "cluster/{name}",
 
+	// Required to allow resetting a member whose database failed to start, e.g. after a failed join.
+	AllowedBeforeInit: true,
+
 	Put: types.EndpointAction{Handler: clusterMemberPut, AccessHandler: access.AllowAuthenticated},
 }
 

internal/rest/resources/control.go

@@ -9,6 +9,7 @@ import (
 	"os"
 	"path/filepath"
 	"slices"
+	"time"
 
 	"github.com/canonical/lxd/shared"
 	"github.com/canonical/lxd/shared/api"
@@ -91,8 +92,14 @@ func controlPost(state types.State, r *http.Request) types.Response {
 			return
 		}
 
+		// Use the daemon's context for cleanup so that it is not tied to the request
+		// context, which may already be expired (e.g. the join timed out), while
+		// still stopping the cleanup if the daemon shuts down.
+		cleanupCtx, cleanupCancel := context.WithTimeout(intState.Context, 2*time.Minute)
+		defer cleanupCancel()
+
 		// Run the pre-remove hook like we do for cluster node removals.
-		err := intState.Hooks.PreRemove(r.Context(), state, true)
+		err := intState.Hooks.PreRemove(cleanupCtx, state, true)
 		if err != nil {
 			logger.Error("Failed to run pre-remove hook on initialization error", slog.String("error", err.Error()))
 		}
@@ -101,6 +108,8 @@ func controlPost(state types.State, r *http.Request) types.Response {
 		// As part of this request the cluster member gets re-executed.
 		// If we don't send the request, re-exec the member manually.
 		if joinInfo == nil || req.JoinToken == "" {
+			// Use the request context, as the re-exec waits for it to be done
+			// so that the response can be sent first.
 			reExec, err := resetClusterMember(r.Context(), state, true)
 			if err != nil {
 				logger.Error("Failed to reset cluster member on bootstrap error", slog.String("error", err.Error()))
@@ -114,13 +123,15 @@ func controlPost(state types.State, r *http.Request) types.Response {
 		}
 
 		url := api.NewURL().Scheme("https").Host(joinInfo.TrustedMember.Address.String())
-		cert, err := shared.GetRemoteCertificate(r.Context(), url.String(), "")
+		cert, err := shared.GetRemoteCertificate(cleanupCtx, url.String(), "")
 		if err != nil {
+			logger.Error("Failed to get certificate of cluster member for cleanup", slog.String("address", url.String()), slog.String("error", err.Error()))
 			return
 		}
 
 		client, err := state.Connect().Member(&url.URL, false, cert)
 		if err != nil {
+			logger.Error("Failed to create client for cluster member cleanup", slog.String("address", url.String()), slog.String("error", err.Error()))
 			return
 		}