internal/db: Close accept channel before closing dqlite on failed bootstrap or join

example/test/main.sh

@@ -1047,6 +1047,66 @@ 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
+}
+
 # allow for running a specific set of tests
 TEST_RESULT="success"
 if [ "${1:-"all"}" = "all" ] || [ "${1}" = "" ]; then
@@ -1061,6 +1121,7 @@ if [ "${1:-"all"}" = "all" ] || [ "${1}" = "" ]; then
   run_test truststore_force_removal
   run_test parallel_joins
   run_test self_deletion
+  run_test failed_join
 elif [ "${1}" = "recover" ]; then
   run_test recover
 elif [ "${1}" = "tokens" ]; then
@@ -1081,6 +1142,8 @@ 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}" = "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},
 }