fix: stop leaking per-job docker networks (#1124)

Leaked per-job networks each hold a subnet of the daemon's address pool and nothing reclaimed them, so a host eventually fails every job with `all predefined address pools have been fully subnetted`. This is what CI hit in https://gitea.com/gitea/runner/actions/runs/742177.

- teardown no longer loses a container, and its network with it: a failed `ContainerRemove` was reported as success, a container whose id was never learned was skipped, and the daemon's own `AutoRemove` teardown was raced
- the idle cleanup reclaims what teardown cannot: networks carry `com.gitea.runner.uuid`, so a runner only touches its own, and a cutoff keeps a job starting during the pass out of scope
- pull failures reported mid-stream were discarded, surfacing later as a confusing `No such image`; they now propagate, and fall back to a local copy instead of failing the job
- `NetworkCreate` retries pool exhaustion briefly, then says which knobs to turn
- digest-pinned images are not re-pulled, and removal kills first so it never waits out Podman's stop timeout (measured 10.1s → 0.09s per container)

---------

Co-authored-by: bircni <bircni@icloud.com>
Reviewed-on: https://gitea.com/gitea/runner/pulls/1124
Reviewed-by: bircni <bircni@icloud.com>

act/container/docker_run.go

@@ -386,32 +386,71 @@ func (cr *containerReference) find() common.Executor {
 	}
 }
 
-// isContainerGone reports whether a failed remove still left the container gone (NotFound or Conflict).
-func isContainerGone(err error) bool {
-	return cerrdefs.IsNotFound(err) || cerrdefs.IsConflict(err)
-}
-
 func (cr *containerReference) remove() common.Executor {
 	return func(ctx context.Context) error {
-		if cr.id == "" {
+		idOrName := cr.id
+		if idOrName == "" && cr.input != nil {
+			idOrName = cr.input.Name
+		}
+		if idOrName == "" {
 			return nil
 		}
 
 		logger := common.Logger(ctx)
-		_, err := cr.cli.ContainerRemove(ctx, cr.id, client.ContainerRemoveOptions{
+		// Kill first so removal never waits out a daemon's stop timeout: Docker kills outright
+		// on a forced remove, Podman sends SIGTERM and waits. Only worth it for a container
+		// this started, and removal can still deal with one it could not kill.
+		if cr.id != "" {
+			_, err := cr.cli.ContainerKill(ctx, cr.id, client.ContainerKillOptions{Signal: "SIGKILL"})
+			if err != nil && !cerrdefs.IsConflict(err) && !cerrdefs.IsNotFound(err) {
+				logger.Debugf("Container %s could not be killed: %v", cr.id, err)
+			}
+		}
+		_, err := cr.cli.ContainerRemove(ctx, idOrName, client.ContainerRemoveOptions{
 			RemoveVolumes: true,
 			Force:         true,
 		})
-		if err != nil && !isContainerGone(err) {
-			logger.Error(fmt.Errorf("failed to remove container: %w", err))
+		switch {
+		case cerrdefs.IsConflict(err):
+			// the daemon's own AutoRemove teardown is running, and it releases the volume
+			// references and the network endpoint only once it finishes
+			cr.waitForRemoval(ctx, idOrName)
+		case err != nil && !cerrdefs.IsNotFound(err):
+			logger.Error(fmt.Errorf("failed to remove container %s: %w", idOrName, err))
+			return nil // keep the id, the container is still there for a later Remove()
 		}
 
-		logger.Debugf("Removed container: %v", cr.id)
+		logger.Debugf("Removed container: %v", idOrName)
 		cr.id = ""
 		return nil
 	}
 }
 
+func (cr *containerReference) waitForRemoval(ctx context.Context, idOrName string) {
+	// per container, against the one minute the post-job executor allows for the whole
+	// cleanup, so a job with several services can spend most of that budget here
+	ctx, cancel := context.WithTimeout(ctx, 15*time.Second)
+	defer cancel()
+
+	waitResult := cr.cli.ContainerWait(ctx, idOrName, client.ContainerWaitOptions{
+		Condition: container.WaitConditionRemoved,
+	})
+	select {
+	case <-waitResult.Result:
+	case <-waitResult.Error:
+	case <-ctx.Done():
+		// the client delivers the result over an unbuffered channel, so leave a receiver
+		// behind or its goroutine parks on the send for the lifetime of the process
+		go func() {
+			select {
+			case <-waitResult.Result:
+			case <-waitResult.Error:
+			}
+		}()
+		common.Logger(ctx).Warnf("Timed out waiting for the daemon to remove container %s, its volumes and network may be left behind", idOrName)
+	}
+}
+
 func (cr *containerReference) mergeContainerConfigs(ctx context.Context, config *container.Config, hostConfig *container.HostConfig) (*container.Config, *container.HostConfig, error) {
 	logger := common.Logger(ctx)
 	input := cr.input