From bc5d29d5cc837cb0d952a7ba239fe7fd4bd0142f Mon Sep 17 00:00:00 2001 From: Claudiu Belu Date: Mon, 19 Oct 2026 13:35:08 +0000 Subject: [PATCH 1/3] internal/db: Close accept channel before closing dqlite on failed bootstrap or join When Bootstrap or Join fails, the reverter closes the dqlite app but keeps the accept channel open. The go-dqlite reader ranging over it keeps receiving connections from the remaining cluster members and panics when it cannot dial the closed app's local socket, crashing the daemon before the join failure cleanup can run. Rotate the accept channel before closing the dqlite app, like Restart already does. Signed-off-by: Claudiu Belu --- internal/db/dqlite.go | 38 +++++++++++++++++++++++++------------- 1 file changed, 25 insertions(+), 13 deletions(-) diff --git a/internal/db/dqlite.go b/internal/db/dqlite.go index 835f5ca6..c772ef16 100644 --- a/internal/db/dqlite.go +++ b/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() From baa58a8ef7d143c4d4b02c90f430841a82fee174 Mon Sep 17 00:00:00 2001 From: Claudiu Belu Date: Mon, 19 Oct 2026 13:37:39 +0000 Subject: [PATCH 2/3] internal/rest/resources: Allow resetting a member whose database is offline On a failed join, the cluster member handling the cleanup resets the joiner through PUT /cluster/{name}. That endpoint required an open database, which a joiner whose dqlite join failed never has, so the reset was refused and the joiner could not be initialized again until its daemon was restarted. Signed-off-by: Claudiu Belu --- internal/rest/resources/cluster.go | 3 +++ 1 file changed, 3 insertions(+) diff --git a/internal/rest/resources/cluster.go b/internal/rest/resources/cluster.go index 691e6ab2..7dea2ede 100644 --- a/internal/rest/resources/cluster.go +++ b/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}, } From 5e5274a193c70be8f87b9e932a6e42fd2881ee8d Mon Sep 17 00:00:00 2001 From: Claudiu Belu Date: Mon, 19 Oct 2026 13:49:39 +0000 Subject: [PATCH 3/3] example: Add failed join system test Signed-off-by: Claudiu Belu --- example/test/main.sh | 63 ++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 63 insertions(+) diff --git a/example/test/main.sh b/example/test/main.sh index a89a7a67..e9d1e46a 100755 --- a/example/test/main.sh +++ b/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