fix: remove the job network even when container removal overruns (#1318)

Job teardown ran every step on one shared minute. When removing the job container took longer, the network removal ran on an expired context and the network stayed behind until the orphan reclaim, which only takes networks older than `workdir_cleanup_age`, 24h by default. All that time it held a subnet of the daemon's address pool.

The job and service containers are now removed in parallel, and the docker proxy sweep, the network and the volumes get a fresh budget after them. The idle reclaim takes unused networks and volumes once they are 10 minutes old, and network removal treats a network that is already gone as removed.

Reviewed-on: https://gitea.com/gitea/runner/pulls/1318
Reviewed-by: bircni <[email protected]>
Co-authored-by: silverwind <[email protected]>

act/container/docker_network.go

@@ -53,7 +53,9 @@ func removeOrphanNetworks(ctx context.Context, cli client.APIClient, runnerUUID,
 networks:
 	for _, n := range networks.Items {
 		result, err := cli.NetworkInspect(ctx, n.ID, client.NetworkInspectOptions{})
-		if err != nil {
+		if cerrdefs.IsNotFound(err) {
+			continue
+		} else if err != nil {
 			errs = append(errs, fmt.Errorf("failed to inspect network %s: %w", n.Name, err))
 			continue
 		}
@@ -71,7 +73,9 @@ networks:
 				continue networks
 			}
 		}
-		if _, err := cli.NetworkRemove(ctx, n.ID, client.NetworkRemoveOptions{}); err != nil {
+		if _, err := cli.NetworkRemove(ctx, n.ID, client.NetworkRemoveOptions{}); cerrdefs.IsNotFound(err) {
+			continue
+		} else if err != nil {
 			errs = append(errs, fmt.Errorf("failed to remove network %s: %w", n.Name, err))
 			continue
 		}
@@ -203,36 +207,35 @@ func NewDockerNetworkRemoveExecutor(name string) common.Executor {
 			return err
 		}
 		defer cli.Close()
+		return removeNetwork(ctx, cli, name)
+	}
+}
 
-		// Make sure that all network of the specified name are removed
-		// cli.NetworkRemove refuses to remove a network if there are duplicates
-		networks, err := cli.NetworkList(ctx, client.NetworkListOptions{})
-		if err != nil {
+func removeNetwork(ctx context.Context, cli client.APIClient, name string) error {
+	networks, err := cli.NetworkList(ctx, client.NetworkListOptions{}) // NetworkRemove by name refuses duplicates, so remove each by ID
+	if err != nil {
+		return err
+	}
+	var errs []error
+	for _, n := range networks.Items {
+		if n.Name != name {
+			continue
+		}
+		result, err := cli.NetworkInspect(ctx, n.ID, client.NetworkInspectOptions{})
+		if cerrdefs.IsNotFound(err) {
+			continue
+		} else if err != nil {
 			return err
 		}
-		// For Gitea, reduce log noise
-		// common.Logger(ctx).Debugf("%v", networks)
-		var errs []error
-		for _, n := range networks.Items {
-			if n.Name == name {
-				result, err := cli.NetworkInspect(ctx, n.ID, client.NetworkInspectOptions{})
-				if err != nil {
-					return err
-				}
-
-				// it holds a subnet out of the daemon's pool until something reclaims it
-				if len(result.Network.Containers) != 0 {
-					common.Logger(ctx).Warnf("Refusing to remove network %s because it still has active endpoints, the idle cleanup reclaims it once they are gone", name)
-					continue
-				}
-				if _, err = cli.NetworkRemove(ctx, n.ID, client.NetworkRemoveOptions{}); err != nil {
-					errs = append(errs, fmt.Errorf("failed to remove network %s: %w", name, err))
-				}
-			}
+		if len(result.Network.Containers) != 0 {
+			common.Logger(ctx).Warnf("Refusing to remove network %s because it still has active endpoints, the orphan cleanup reclaims it once they are gone", name)
+			continue
+		}
+		if _, err = cli.NetworkRemove(ctx, n.ID, client.NetworkRemoveOptions{}); err != nil && !cerrdefs.IsNotFound(err) {
+			errs = append(errs, fmt.Errorf("failed to remove network %s: %w", name, err))
 		}
-
-		return errors.Join(errs...)
 	}
+	return errors.Join(errs...)
 }
 
 // IsolatedCacheContainer returns the container holding addr, when jobs on jobNetwork cannot reach it there.

act/container/docker_network_test.go

@@ -116,7 +116,7 @@ func TestConnectAndDisconnectNetwork(t *testing.T) {
 // Of this runner's networks, only the ones nothing but its own cache container is attached to
 // and old enough to predate any job now starting are the runner's to reclaim. An unexpected
 // NetworkRemove fails the test on its own, since testify has no expectation to match it against.
-func TestRemoveOrphanNetworks(t *testing.T) {
+func TestRemoveNetworks(t *testing.T) {
 	ctx := t.Context()
 	cutoff := time.Date(2026, time.April, 29, 20, 0, 0, 0, time.UTC)
 	client := &mockDockerClient{}
@@ -128,11 +128,13 @@ func TestRemoveOrphanNetworks(t *testing.T) {
 		{ID: "starting"},
 		{ID: "cached"},
 		{ID: "shared"},
+		{ID: "gone"},
+		{ID: "raced"},
 	}}, nil)
 	client.On("NetworkInspect", ctx, "orphan", mobyclient.NetworkInspectOptions{}).
-		Return(mobyclient.NetworkInspectResult{}, nil)
+		Return(mobyclient.NetworkInspectResult{}, nil).Twice()
 	client.On("NetworkInspect", ctx, "busy", mobyclient.NetworkInspectOptions{}).
-		Return(mobyclient.NetworkInspectResult{Network: network.Inspect{Containers: map[string]network.EndpointResource{"c": {}}}}, nil)
+		Return(mobyclient.NetworkInspectResult{Network: network.Inspect{Containers: map[string]network.EndpointResource{"c": {}}}}, nil).Twice()
 	client.On("NetworkInspect", ctx, "starting", mobyclient.NetworkInspectOptions{}).
 		Return(mobyclient.NetworkInspectResult{Network: network.Inspect{
 			Network:    network.Network{Created: cutoff.Add(time.Second)},
@@ -142,13 +144,28 @@ func TestRemoveOrphanNetworks(t *testing.T) {
 		Return(mobyclient.NetworkInspectResult{Network: network.Inspect{Containers: map[string]network.EndpointResource{"0123456789abcdef": {}}}}, nil)
 	client.On("NetworkInspect", ctx, "shared", mobyclient.NetworkInspectOptions{}).
 		Return(mobyclient.NetworkInspectResult{Network: network.Inspect{Containers: map[string]network.EndpointResource{"0123456789abcdef": {}, "job": {}}}}, nil)
+	client.On("NetworkInspect", ctx, "gone", mobyclient.NetworkInspectOptions{}).
+		Return(mobyclient.NetworkInspectResult{}, cerrdefs.ErrNotFound).Twice()
+	client.On("NetworkInspect", ctx, "raced", mobyclient.NetworkInspectOptions{}).
+		Return(mobyclient.NetworkInspectResult{}, nil).Twice()
+	client.On("NetworkRemove", ctx, "raced", mobyclient.NetworkRemoveOptions{}).
+		Return(mobyclient.NetworkRemoveResult{}, cerrdefs.ErrNotFound).Twice()
 	client.On("NetworkRemove", ctx, "orphan", mobyclient.NetworkRemoveOptions{}).
-		Return(mobyclient.NetworkRemoveResult{}, nil)
+		Return(mobyclient.NetworkRemoveResult{}, nil).Twice()
 	client.On("NetworkDisconnect", ctx, "cached", mobyclient.NetworkDisconnectOptions{Container: "0123456789abcdef", Force: true}).
 		Return(nil).Once()
 	client.On("NetworkRemove", ctx, "cached", mobyclient.NetworkRemoveOptions{}).
 		Return(mobyclient.NetworkRemoveResult{}, nil).Once()
 
 	require.NoError(t, removeOrphanNetworks(ctx, client, "runner-1", "0123456789ab", cutoff))
+
+	client.On("NetworkList", ctx, mobyclient.NetworkListOptions{}).Return(mobyclient.NetworkListResult{Items: []network.Summary{
+		{ID: "orphan", Name: "job"},
+		{ID: "busy", Name: "job"},
+		{ID: "gone", Name: "job"},
+		{ID: "raced", Name: "job"},
+		{ID: "shared", Name: "other"},
+	}}, nil)
+	require.NoError(t, removeNetwork(ctx, client, "job"))
 	client.AssertExpectations(t)
 }

act/runner/run_context.go

@@ -22,6 +22,7 @@ import (
 	"path/filepath"
 	"regexp"
 	"runtime"
+	"runtime/debug"
 	"slices"
 	"strconv"
 	"strings"
@@ -719,31 +720,55 @@ func (rc *RunContext) cleanupJobResources(networkName string, createAndDeleteNet
 		logger := common.Logger(ctx)
 		rc.printServiceLogs(ctx)
 		errs := []error{rc.closeDockerProxy(ctx)}
+		var containers []common.Executor
 		if rc.JobContainer != nil {
-			errs = append(errs, rc.JobContainer.Remove()(ctx))
+			containers = append(containers, rc.JobContainer.Remove())
 		}
 		if len(rc.serviceContainers) > 0 {
 			logger.Infof("Cleaning up services for job %s", rc.JobName)
-			errs = append(errs, rc.stopServiceContainers()(ctx))
+			containers = append(containers, rc.stopServiceContainers())
 		}
+		errs = append(errs, runAll(ctx, containers))
+		ctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), jobCleanupTimeout)
+		defer cancel()
 		if !common.Dryrun(ctx) && !rc.Config.NoSweep && (preclean || rc.hadDockerProxy) {
-			errs = append(errs, container.RemoveDockerJobResources(ctx, rc.jobContainerName()))
+			errs = append(errs, container.RemoveDockerJobResources(ctx, rc.jobContainerName())) // after the containers, whose endpoints would block removing the job networks
 		}
+		var rest []common.Executor
 		if preclean || rc.deferVolumeCleanup == nil {
-			errs = append(errs, rc.cleanupJobVolumes(ctx))
+			rest = append(rest, rc.cleanupJobVolumes)
 		} else {
 			rc.deferVolumeCleanup(rc.cleanupJobVolumes)
 		}
 		if createAndDeleteNetwork {
 			logger.Infof("Cleaning up network for job %s, and network name is: %s", rc.JobName, networkName)
-			errs = append(errs, container.NewDockerNetworkDisconnectExecutor(networkName, rc.Config.CacheContainer).
-				IfBool(rc.Config.CacheContainer != "")(ctx))
-			errs = append(errs, container.NewDockerNetworkRemoveExecutor(networkName)(ctx))
+			rest = append(rest, func(ctx context.Context) error {
+				return errors.Join(
+					container.NewDockerNetworkDisconnectExecutor(networkName, rc.Config.CacheContainer).IfBool(rc.Config.CacheContainer != "")(ctx),
+					container.NewDockerNetworkRemoveExecutor(networkName)(ctx))
+			})
 		}
-		return errors.Join(errs...)
+		return errors.Join(append(errs, runAll(ctx, rest))...)
 	}
 }
 
+func runAll(ctx context.Context, executors []common.Executor) error {
+	errs := make([]error, len(executors))
+	var wg sync.WaitGroup
+	for i, executor := range executors {
+		wg.Go(func() {
+			defer func() {
+				if recovered := recover(); recovered != nil {
+					errs[i] = fmt.Errorf("panic: %v\n%s", recovered, debug.Stack())
+				}
+			}()
+			errs[i] = executor(ctx)
+		})
+	}
+	wg.Wait()
+	return errors.Join(errs...)
+}
+
 func (rc *RunContext) createJobVolumes(ctx context.Context, mounts map[string]string) error {
 	runnerUUID := rc.Config.ContainerNetworkCreateOptions.RunnerUUID
 	if runnerUUID == "" || common.Dryrun(ctx) {

act/runner/run_context_test.go

@@ -20,6 +20,7 @@ import (
 	"runtime"
 	"slices"
 	"strings"
+	"sync"
 	"testing"
 	"testing/synctest"
 	"time"
@@ -775,10 +776,10 @@ func TestRemoveDockerActionImagesByTag(t *testing.T) {
 	assert.Equal(t, []string{"act-action-dockeraction:latest-job"}, removed)
 }
 
-func TestCleanupJobResourcesContinuesAfterFailure(t *testing.T) {
+func TestCleanupJobResources(t *testing.T) {
 	t.Setenv("TMPDIR", "/tmp")
 	proxyDir := t.TempDir()
-	for _, name := range []string{"synchronous", "proxy", "closed proxy", "kept proxy", "deferred", "preclean deferred", "canceled"} {
+	for _, name := range []string{"synchronous", "proxy", "closed proxy", "kept proxy", "deferred", "preclean deferred", "canceled proxy"} {
 		t.Run(name, func(t *testing.T) {
 			proxy, preclean, deferred := strings.HasSuffix(name, "proxy"), strings.HasPrefix(name, "preclean"), strings.HasSuffix(name, "deferred")
 			removesLabelled := (proxy || preclean) && name != "kept proxy"
@@ -787,8 +788,20 @@ func TestCleanupJobResourcesContinuesAfterFailure(t *testing.T) {
 			}
 			jobError, removeError, closeError := errors.New("remove job"), errors.New("remove service"), errors.New("close service")
 			job, service := &containerMock{}, &containerMock{}
-			job.On("Remove").Return(func(context.Context) error { return jobError }).Once()
-			service.On("Remove").Return(func(context.Context) error { return removeError }).Once()
+			removeAlongside := func(started, other chan struct{}, failure error) func(context.Context) error {
+				return func(context.Context) error {
+					close(started)
+					select {
+					case <-other:
+					case <-time.After(10 * time.Second):
+						t.Error("the containers are removed one after the other instead of alongside each other")
+					}
+					return failure
+				}
+			}
+			jobRemoving, serviceRemoving := make(chan struct{}), make(chan struct{})
+			job.On("Remove").Return(removeAlongside(jobRemoving, serviceRemoving, jobError)).Once()
+			service.On("Remove").Return(removeAlongside(serviceRemoving, jobRemoving, removeError)).Once()
 			service.On("Close").Return(func(context.Context) error { return closeError }).Once()
 			rc := &RunContext{
 				Config:            &Config{CacheContainer: "cache-container", NoSweep: name == "kept proxy"},
@@ -825,7 +838,7 @@ func TestCleanupJobResourcesContinuesAfterFailure(t *testing.T) {
 			}
 			ctx, cancel := context.WithCancel(t.Context())
 			defer cancel()
-			if name == "canceled" {
+			if name == "canceled proxy" {
 				cancel()
 			}
 			err := rc.cleanupJobResources("job-network", true, preclean)(ctx)
@@ -839,14 +852,10 @@ func TestCleanupJobResourcesContinuesAfterFailure(t *testing.T) {
 			for _, failure := range []error{jobError, removeError, closeError} {
 				require.ErrorIs(t, err, failure)
 			}
-			if name == "canceled" {
-				require.ErrorIs(t, err, context.Canceled)
-			} else {
-				for _, operation := range []string{"DELETE /volumes/" + rc.jobContainerName(), "DELETE /volumes/" + rc.jobContainerName() + "-env", "POST /networks/job-network/disconnect", "GET /networks"} {
-					require.ErrorContains(t, err, operation)
-				}
-				assert.Equal(t, 2, volumeRemovals)
+			for _, operation := range []string{"DELETE /volumes/" + rc.jobContainerName(), "DELETE /volumes/" + rc.jobContainerName() + "-env", "POST /networks/job-network/disconnect", "GET /networks"} {
+				require.ErrorContains(t, err, operation)
 			}
+			assert.Equal(t, 2, volumeRemovals)
 			assert.Equal(t, removesLabelled, strings.Contains(err.Error(), "labelled GET /containers/json"))
 			if removesLabelled {
 				require.ErrorContains(t, err, "labelled GET /networks")
@@ -1360,8 +1369,11 @@ func TestRunContext_cleanupFailedStart(t *testing.T) {
 	} {
 		t.Run(testcase.name, func(t *testing.T) {
 			var operations []string
+			var operationsMu sync.Mutex
 			record := func(operation string, failure error) func(context.Context) error {
 				return func(context.Context) error {
+					operationsMu.Lock()
+					defer operationsMu.Unlock()
 					operations = append(operations, operation)
 					return failure
 				}
@@ -1403,7 +1415,7 @@ func TestRunContext_cleanupFailedStart(t *testing.T) {
 			ctx := common.WithLogger(common.WithDryrun(t.Context(), true), logger)
 			rc.ExprEval = rc.NewExpressionEvaluator(ctx)
 			err = rc.startContainer().Then(rc.stopContainer()).Finally(rc.closeContainer())(ctx)
-			want := "job.Remove service.Remove service.Close service.Pull job.Pull"
+			want := "job.Remove service.Close service.Remove service.Pull job.Pull"
 			switch {
 			case testcase.pullError != nil:
 				require.ErrorIs(t, err, testcase.pullError)
@@ -1418,7 +1430,9 @@ func TestRunContext_cleanupFailedStart(t *testing.T) {
 					require.NoError(t, err)
 				}
 			}
-			assert.Equal(t, strings.Fields(want+" job.Remove service.Remove service.Close job.Close"), operations)
+			slices.Sort(operations[:3])
+			slices.Sort(operations[len(operations)-4 : len(operations)-1])
+			assert.Equal(t, strings.Fields(want+" job.Remove service.Close service.Remove job.Close"), operations)
 			assert.Empty(t, rc.Masks)
 			assert.Equal(t, testcase.pullError == nil && testcase.health != container.HealthUnhealthy, slices.ContainsFunc(hook.AllRecords(), func(entry logging.Record) bool {
 				return strings.HasPrefix(entry.Message, "::group::Print service container logs")