tests: wait for the mock daemons to be ready before running

`TestStartAndStopDaemon` regularly failed in CI on authorization errors:
the tests connected to a bus that wasn't listening yet.

`runDaemons` started thirteen containers in parallel and gave them a
fixed five seconds to come up. That is plenty on an idle machine, where
they are ready in under a second, but nothing guarantees it on a loaded
runner.

Poll for actual readiness instead: both bus sockets must accept
connections and polkitd must have claimed its name on the system bus,
as the containers start it after the buses. This is both faster in the
common case and reliable under load.

While at it, guard `errsDocker` with a mutex. The container goroutines
outlive `runDaemons` and kept writing to it while the main goroutine
read it, which the race detector flagged.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Author: denisonbarbosa

cmd/adsysd/integration_tests/adsys_test.go

@@ -13,12 +13,14 @@ import (
 	"os"
 	"os/exec"
 	"path/filepath"
+	"slices"
 	"strconv"
 	"strings"
 	"sync"
 	"testing"
 	"time"
 
+	"github.com/godbus/dbus/v5"
 	"github.com/stretchr/testify/require"
 	"github.com/ubuntu/adsys"
 	"github.com/ubuntu/adsys/cmd/adsysd/client"
@@ -30,6 +32,11 @@ import (
 
 const dockerSystemDaemonsImage = "ghcr.io/ubuntu/adsys/systemdaemons:0.1"
 
+// daemonsStartupTimeout is how long we wait for the mock containers to expose
+// their buses. It is deliberately generous: the containers are started in
+// parallel and a loaded CI runner can take a while to get them all up.
+const daemonsStartupTimeout = 90 * time.Second
+
 var (
 	rootProjectDir string
 )
@@ -437,18 +444,38 @@ func runDaemons() (teardown func()) {
 	groupFile := writeUserDatabase(dir, "group", "/etc/group")
 
 	var errsDocker error
-	var wg sync.WaitGroup
+	var errsDockerMu sync.Mutex
+	// The container goroutines outlive this function, so reads and writes of
+	// errsDocker must be synchronized.
+	addDockerErr := func(err error) {
+		errsDockerMu.Lock()
+		defer errsDockerMu.Unlock()
+		errsDocker = errors.Join(errsDocker, err)
+	}
+	dockerErr := func() error {
+		errsDockerMu.Lock()
+		defer errsDockerMu.Unlock()
+		return errsDocker
+	}
+
+	var runWG sync.WaitGroup
+	runDone := make(map[string]chan struct{}, len(answers))
+	for answer := range answers {
+		runDone[answer] = make(chan struct{})
+	}
 	for answer, socketDir := range answers {
-		wg.Add(1)
+		runWG.Add(1)
 
-		go func() {
-			defer wg.Done()
+		go func(answer, socketDir string, done chan struct{}) {
+			defer runWG.Done()
+			defer close(done)
 
 			// 24.04 introduced some polkit changes that make the daemon drop root privileges before executing. In order
 			// to be able to connect to the bus and run polkitd, we need more permissions in the socket directory.
 			//nolint:gosec
 			if err := os.MkdirAll(socketDir, 0755); err != nil {
-				log.Fatalf("Setup: can’t create %s socket directory: %v", answer, err)
+				addDockerErr(fmt.Errorf("Setup: can’t create %s socket directory: %w", answer, err))
+				return
 			}
 
 			// #nosec G204: we control the name in tests
@@ -462,50 +489,189 @@ func runDaemons() (teardown func()) {
 				dockerSystemDaemonsImage,
 				answer,
 			)
-			out, _ := cmd.CombinedOutput()
+			out, err := cmd.CombinedOutput()
+			if cmd.ProcessState == nil {
+				if err == nil {
+					err = errors.New("command returned without a process state")
+				}
+				addDockerErr(fmt.Errorf("Error running system daemons container named %q: %w\n%v", answer, err, string(out)))
+				return
+			}
 			// Docker stop -t 0 will kill it anyway the container with exit code 143 or 137 in mantic (-t 0 does not work there)
-			if cmd.ProcessState.ExitCode() > 0 && cmd.ProcessState.ExitCode() != 137 && cmd.ProcessState.ExitCode() != 143 {
-				errsDocker = errors.Join(errsDocker, fmt.Errorf("Error running system daemons container named %q:\nExit code: %d\n%v", answer, cmd.ProcessState.ExitCode(), string(out)))
+			if exitCode := cmd.ProcessState.ExitCode(); exitCode > 0 && exitCode != 137 && exitCode != 143 {
+				addDockerErr(fmt.Errorf("Error running system daemons container named %q:\nExit code: %d\n%v", answer, exitCode, string(out)))
 			}
-		}()
+		}(answer, socketDir, runDone[answer])
 	}
 
 	for a, s := range answers {
 		dbusSockets[a] = fmt.Sprintf("unix:path=%s", s)
 	}
 
-	// give time for polkit containers to start
-	// TODO: wait for polkit containers to be ready
-	time.Sleep(5 * time.Second)
+	// Wait for the polkit containers to be ready instead of hoping a fixed delay
+	// is enough: a loaded machine may need much longer, and connecting to a bus
+	// that is not listening yet fails the tests using it.
+	stopContainers := func() error {
+		var stopWG sync.WaitGroup
+		stopErrors := make(chan error, len(answers))
+		for answer := range answers {
+			stopWG.Add(1)
+			go func(answer string) {
+				defer stopWG.Done()
+
+				deadline := time.Now().Add(daemonsStartupTimeout)
+				var lastErr error
+				var lastOut string
+				for {
+					select {
+					case <-runDone[answer]:
+						return
+					default:
+					}
+					if time.Now().After(deadline) {
+						stopErr := fmt.Errorf("Teardown: can't stop system daemons container %q within %s", answer, daemonsStartupTimeout)
+						if lastErr != nil {
+							stopErr = fmt.Errorf("%w (last error: %w: %s)", stopErr, lastErr, lastOut)
+						}
+						stopErrors <- stopErr
+						return
+					}
+
+					ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+					// #nosec G204: we control the args in tests
+					out, err := exec.CommandContext(ctx, "docker", "stop", "-t", "0", containerName+answer).CombinedOutput()
+					cancel()
+					if err != nil {
+						lastErr = err
+						lastOut = string(out)
+					}
+					select {
+					case <-runDone[answer]:
+						return
+					case <-time.After(100 * time.Millisecond):
+					}
+				}
+			}(answer)
+		}
+
+		stopWG.Wait()
+		close(stopErrors)
+		var stopErr error
+		for err := range stopErrors {
+			stopErr = errors.Join(stopErr, err)
+		}
+		if stopErr != nil {
+			return stopErr
+		}
+
+		runWG.Wait()
+		return dockerErr()
+	}
+	cleanup := func() error {
+		stopErr := stopContainers()
+		if stopErr != nil {
+			stopErr = fmt.Errorf("can't stop system daemons containers: %w", stopErr)
+		}
+		if err := os.RemoveAll(dir); err != nil {
+			err = fmt.Errorf("failed to delete temporary directory: %w", err)
+			return errors.Join(stopErr, err)
+		}
+		return stopErr
+	}
 
-	// Check if we got startup docker error
-	if errsDocker != nil {
-		log.Fatalf("Setup: docker errors (some may still be ALIVE): %v", errsDocker)
+	if err := waitForDaemons(answers, dockerErr); err != nil {
+		if cleanupErr := cleanup(); cleanupErr != nil {
+			err = errors.Join(err, fmt.Errorf("cleanup after setup failure: %w", cleanupErr))
+		}
+		log.Fatalf("Setup: %v", err)
 	}
 
 	return func() {
-		defer func() {
-			err := os.RemoveAll(dir)
-			if err != nil {
-				log.Fatalf("Teardown: failed to delete temporary directory: %v", err)
+		if err := cleanup(); err != nil {
+			log.Fatalf("Teardown: %v", err)
+		}
+	}
+}
+
+// waitForDaemons blocks until every mock container is ready to serve requests,
+// or until daemonsStartupTimeout elapses. It fails early if a container reported
+// a startup error in the meantime.
+func waitForDaemons(answers map[string]string, dockerErr func() error) error {
+	deadline := time.Now().Add(daemonsStartupTimeout)
+	ctx, cancel := context.WithDeadline(context.Background(), deadline)
+	defer cancel()
+
+	for answer, socketDir := range answers {
+		for {
+			err := daemonsReady(ctx, socketDir)
+			if err == nil {
+				break
 			}
-		}()
 
-		for answer := range answers {
-			go func() {
-				// #nosec G204: we control the args in tests
-				out, err := exec.Command("docker", "stop", "-t", "0", containerName+answer).CombinedOutput()
-				if err != nil {
-					errsDocker = errors.Join(errsDocker, fmt.Errorf("Teardown: can't stop system daemons container: %v", string(out)))
-				}
-			}()
+			if dockerErrs := dockerErr(); dockerErrs != nil {
+				return fmt.Errorf("docker errors while waiting for the %q daemons: %w", answer, dockerErrs)
+			}
+			if time.Now().After(deadline) {
+				return fmt.Errorf("%q daemons not ready after %s: %w", answer, daemonsStartupTimeout, err)
+			}
+
+			time.Sleep(100 * time.Millisecond)
 		}
+	}
 
-		wg.Wait()
-		if errsDocker != nil {
-			log.Fatalf("Teardown: docker errors: %v", errsDocker)
+	if dockerErrs := dockerErr(); dockerErrs != nil {
+		return fmt.Errorf("docker errors while waiting for daemon readiness: %w", dockerErrs)
+	}
+
+	return nil
+}
+
+// daemonsReady reports whether the mock daemons exposed in socketDir can serve
+// requests: both buses must accept connections and polkitd must have claimed its
+// name on the system bus, as it is started after them.
+func daemonsReady(ctx context.Context, socketDir string) error {
+	systemSocket := filepath.Join(socketDir, "system_bus_socket")
+	for _, socket := range []string{systemSocket, filepath.Join(socketDir, "session_bus_socket")} {
+		conn, err := (&net.Dialer{}).DialContext(ctx, "unix", socket)
+		if err != nil {
+			return err
 		}
+		_ = conn.Close()
 	}
+
+	netConn, err := (&net.Dialer{}).DialContext(ctx, "unix", systemSocket)
+	if err != nil {
+		return err
+	}
+	if deadline, ok := ctx.Deadline(); ok {
+		if err := netConn.SetDeadline(deadline); err != nil {
+			_ = netConn.Close()
+			return fmt.Errorf("set D-Bus readiness deadline: %w", err)
+		}
+	}
+
+	conn, err := dbus.NewConn(netConn, dbus.WithContext(ctx))
+	if err != nil {
+		_ = netConn.Close()
+		return fmt.Errorf("create D-Bus connection: %w", err)
+	}
+	defer func() { _ = conn.Close() }()
+	if err := conn.Auth(nil); err != nil {
+		return fmt.Errorf("authenticate D-Bus connection: %w", err)
+	}
+	if err := conn.Hello(); err != nil {
+		return fmt.Errorf("complete D-Bus handshake: %w", err)
+	}
+
+	var names []string
+	if err := conn.BusObject().CallWithContext(ctx, "org.freedesktop.DBus.ListNames", 0).Store(&names); err != nil {
+		return err
+	}
+	if !slices.Contains(names, "org.freedesktop.PolicyKit1") {
+		return errors.New("polkitd has not claimed its name on the system bus yet")
+	}
+
+	return nil
 }
 
 // writeUserDatabase copies the host NSS database file (/etc/passwd or /etc/group)

internal/policies/scripts/export_test.go

@@ -3,6 +3,7 @@ package scripts
 import (
 	"context"
 	"os/user"
+	"time"
 )
 
 const (
@@ -21,3 +22,9 @@ func WithUserLookup(userLookup func(string) (*user.User, error)) Option {
 func RunScriptsWithBusyScriptRetryHook(ctx context.Context, order string, allowOrderMissing bool, hook func()) error {
 	return runScripts(ctx, order, allowOrderMissing, hook)
 }
+
+// RunScriptsWithBusyScriptRetrySettings executes scripts with explicit ETXTBSY
+// retry-loop settings and calls hook when a retry is about to happen.
+func RunScriptsWithBusyScriptRetrySettings(ctx context.Context, order string, allowOrderMissing bool, retries int, delay time.Duration, hook func()) error {
+	return runScriptsWithBusyScriptRetrySettings(ctx, order, allowOrderMissing, retries, delay, hook)
+}

internal/policies/scripts/scripts.go

@@ -239,6 +239,10 @@ func RunScripts(ctx context.Context, order string, allowOrderMissing bool) error
 }
 
 func runScripts(ctx context.Context, order string, allowOrderMissing bool, onBusyScriptRetry func()) (err error) {
+	return runScriptsWithBusyScriptRetrySettings(ctx, order, allowOrderMissing, busyScriptRetries, busyScriptDelay, onBusyScriptRetry)
+}
+
+func runScriptsWithBusyScriptRetrySettings(ctx context.Context, order string, allowOrderMissing bool, retries int, delay time.Duration, onBusyScriptRetry func()) (err error) {
 	defer decorate.OnError(&err, gotext.Get("can't run scripts listed in %s", order))
 
 	log.Infof(ctx, "Calling RunScripts on %q", order)
@@ -298,7 +302,7 @@ func runScripts(ctx context.Context, order string, allowOrderMissing bool, onBus
 		}
 		script := filepath.Join(baseDir, scriptPath)
 		log.Debugf(ctx, "Running script %q", script)
-		if err := runScript(ctx, script, onBusyScriptRetry); err != nil {
+		if err := runScript(ctx, script, retries, delay, onBusyScriptRetry); err != nil {
 			log.Warningf(ctx, "%q failed to run\n%v", script, err)
 		}
 	}
@@ -308,7 +312,7 @@ func runScripts(ctx context.Context, order string, allowOrderMissing bool, onBus
 
 // runScript executes a single script, retrying while the kernel reports it as
 // still being open for writing (ETXTBSY).
-func runScript(ctx context.Context, script string, onBusyScriptRetry func()) error {
+func runScript(ctx context.Context, script string, retries int, delay time.Duration, onBusyScriptRetry func()) error {
 	for i := 0; ; i++ {
 		// #nosec G204 - this variable is coming from concatenation of an order file.
 		// Permissions are restricted to the owner of the order file, which is the one executing
@@ -318,7 +322,7 @@ func runScript(ctx context.Context, script string, onBusyScriptRetry func()) err
 		cmd.Stderr = os.Stderr
 
 		err := cmd.Run()
-		if !errors.Is(err, syscall.ETXTBSY) || i == busyScriptRetries {
+		if !errors.Is(err, syscall.ETXTBSY) || i == retries {
 			return err
 		}
 
@@ -329,7 +333,7 @@ func runScript(ctx context.Context, script string, onBusyScriptRetry func()) err
 		select {
 		case <-ctx.Done():
 			return err
-		case <-time.After(busyScriptDelay):
+		case <-time.After(delay):
 		}
 	}
 }