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 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() 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}, }