1// SPDX-License-Identifier: MIT
2
3// Package container is the container sandbox, for hosts without KVM: one
4// container per job with its own Docker daemon inside (the sysbox runtime when
5// the daemon lists it, else a privileged container), on its own bridge network
6// and with its own scratch disk. The scratch disk is a sparse ext4 file of
7// ERMITE_RUNNER_JOB_DISK, loop-mounted on the host (so it needs root, like the
8// rest of this mode) and bind-mounted at /scratch: the nested dockerd keeps its
9// data root there and the job workspace is a directory of it, so a job cannot
10// write more than its disk size to the host (the container's own writable layer
11// stays small: nothing big is written outside /scratch and /home/runner/work).
12// Everything Docker-side carries the label sandbox.LabelJob so that Reap finds
13// it after a crash.
14//
15// The executor stream is a Unix socket in a per-job directory of the runner's
16// state directory, bind-mounted into the container. The runner executable is
17// bind-mounted read-only. Both paths are host paths given to the Docker daemon:
18// the daemon must run on the same host (and mount namespace) as the runner.
19package container
20
21import (
22 "context"
23 "errors"
24 "fmt"
25 "net"
26 "net/netip"
27 "os"
28 "path/filepath"
29 "slices"
30 "strings"
31 "sync"
32 "time"
33
34 "github.com/bnema/zerowrap"
35
36 "bnema.dev/ermite-runner/internal/cmdrun"
37 "bnema.dev/ermite-runner/internal/netpolicy"
38 "bnema.dev/ermite-runner/internal/sandbox"
39 "bnema.dev/ermite-runner/internal/sandbox/disk"
40 "bnema.dev/ermite-runner/internal/sandbox/netsetup"
41)
42
43// SysboxRuntime is the name of the sysbox runtime in `docker info`.
44const SysboxRuntime = "sysbox-runc"
45
46// bridgePrefix must match the "erbr*" wildcard of the nftables policy.
47const bridgePrefix = "erbr"
48
49// Options are the testable knobs of the driver.
50type Options struct {
51 Run cmdrun.Runner
52 // Net overrides the network plumbing (tests).
53 Net netsetup.Plumbing
54 // engine overrides the Docker client (tests).
55 engine engine
56 // UnsafeAllowRootlessForTests lets tests exercise the Engine API and the executor against
57 // the rootless daemon of a workstation. Never set outside tests: a nested
58 // dockerd cannot work in a rootless daemon, so the jobs would run without
59 // Docker. The factories of package drivers never set it.
60 UnsafeAllowRootlessForTests bool
61 // ConnectTimeout bounds container start + init + executor connection
62 // (default 90 s).
63 ConnectTimeout time.Duration
64 // Mounter overrides the loop mounts (tests).
65 Mounter *disk.Mounter
66 // FreeSpace overrides statfs for the free-space warning (tests).
67 FreeSpace func(string) (int64, error)
68}
69
70// Driver implements sandbox.Driver for containers.
71type Driver struct {
72 cfg sandbox.Config
73 opt Options
74 eng engine
75 net netsetup.Plumbing
76 mnt disk.Mounter
77 lock *sandbox.HostLock
78 image string
79 runtime string // "" when privileged
80 privileged bool
81 exe string
82
83 mu sync.Mutex
84 slots *sandbox.Slots
85}
86
87// New takes the host-wide lock, connects to the Docker daemon (DOCKER_HOST
88// honoured) and checks it: not rootless, the job image present (pulled if
89// missing), the runtime decided (sysbox when listed; else privileged, unless
90// ERMITE_RUNNER_DENY_PRIVILEGED=1, which refuses). IPv4 forwarding is enabled
91// host-wide and stays on (see the README).
92func New(ctx context.Context, cfg sandbox.Config, opt Options) (*Driver, error) {
93 if opt.Run == nil {
94 opt.Run = cmdrun.Exec{}
95 }
96 if opt.ConnectTimeout == 0 {
97 opt.ConnectTimeout = 90 * time.Second
98 }
99 d := &Driver{cfg: cfg, opt: opt, image: cfg.OCIImage(), slots: sandbox.NewSlots(cfg.Capacity)}
100 if opt.Mounter != nil {
101 d.mnt = *opt.Mounter
102 } else {
103 d.mnt = disk.Mounter{Run: opt.Run}
104 }
105 exe, err := cfg.ExecutablePath()
106 if err != nil {
107 return nil, err
108 }
109 if err := sandbox.CheckStatic(exe); err != nil {
110 return nil, err
111 }
112 d.exe = exe
113
114 if d.lock, err = sandbox.AcquireHostLock(cfg.LockFile); err != nil {
115 return nil, err
116 }
117 eng := opt.engine
118 if eng == nil {
119 e, err := newDockerEngine()
120 if err != nil {
121 _ = d.lock.Release()
122 return nil, err
123 }
124 eng = e
125 }
126 d.eng = eng
127 fail := func(err error) (*Driver, error) {
128 _ = eng.Close()
129 _ = d.lock.Release()
130 return nil, err
131 }
132
133 info, err := eng.Info(ctx)
134 if err != nil {
135 return fail(fmt.Errorf("cannot reach the Docker daemon (DOCKER_HOST): %w", err))
136 }
137 if info.Rootless && !opt.UnsafeAllowRootlessForTests {
138 return fail(errRootless)
139 }
140 if info.OSType != "" && info.OSType != "linux" {
141 return fail(fmt.Errorf("the Docker daemon runs %s containers, need linux", info.OSType))
142 }
143 log := zerowrap.FromCtx(ctx)
144 if slices.Contains(info.Runtimes, SysboxRuntime) {
145 d.runtime = SysboxRuntime
146 log.Info().Str("runtime", SysboxRuntime).Msg("container isolation: using the sysbox runtime")
147 } else {
148 if cfg.DenyPrivileged {
149 return fail(fmt.Errorf("the sysbox runtime is not available and %s=1 forbids privileged containers (install sysbox, or unset the setting to accept host-root-equivalent jobs)", sandbox.EnvDenyPrivileged))
150 }
151 d.privileged = true
152 log.Warn().Msg(privilegedWarning)
153 }
154 if err := disk.CheckMkfs(ctx, opt.Run); err != nil {
155 return fail(err)
156 }
157 if err := os.MkdirAll(cfg.StateDir, 0o755); err != nil {
158 return fail(fmt.Errorf("state directory: %w", err))
159 }
160 sandbox.WarnIfLowSpace(ctx, cfg.StateDir, cfg.Capacity, cfg.JobDiskMiB, opt.FreeSpace)
161 if err := eng.EnsureImage(ctx, d.image); err != nil {
162 return fail(err)
163 }
164 d.net = opt.Net
165 if d.net == nil {
166 d.net = newNet(opt.Run, cfg.DNSPort, info.Podman)
167 }
168 if err := d.net.EnableForwarding(); err != nil {
169 return fail(err)
170 }
171 return d, nil
172}
173
174// newNet builds the host network plumbing. Only podman creates the bridge
175// address late (see daemonInfo.Podman), so only there the DNS listener may come
176// before the address; on Docker a missing gateway address stays a startup error.
177func newNet(run cmdrun.Runner, dnsPort int, podman bool) *netsetup.Net {
178 n := netsetup.New(run, dnsPort, "")
179 n.DNS.FreeBind = podman
180 return n
181}
182
183// maxSockPath is the longest Unix socket path Linux accepts (108 with the NUL).
184const maxSockPath = 107
185
186// privilegedWarning is logged when the driver starts and again at every job
187// start: a privileged container is host-root equivalent.
188const privilegedWarning = "privileged container: host-root equivalent, network policy best effort"
189
190// Mode implements sandbox.Driver.
191func (d *Driver) Mode() sandbox.Mode { return sandbox.Container }
192
193// Isolation implements sandbox.Driver.
194func (d *Driver) Isolation() sandbox.Isolation {
195 if d.privileged {
196 return sandbox.IsolationContainerPrivileged
197 }
198 return sandbox.IsolationContainerSysbox
199}
200
201// Privileged reports whether jobs run privileged (no sysbox).
202func (d *Driver) Privileged() bool { return d.privileged }
203
204// Close implements sandbox.Driver.
205func (d *Driver) Close() error {
206 if d.net != nil {
207 d.net.Close()
208 }
209 err := d.eng.Close()
210 return errors.Join(err, d.lock.Release())
211}
212
213// New implements sandbox.Driver.
214func (d *Driver) New() (sandbox.Sandbox, error) {
215 id, err := sandbox.NewID()
216 if err != nil {
217 return nil, err
218 }
219 d.mu.Lock()
220 defer d.mu.Unlock()
221 slot, err := d.slots.Acquire()
222 if err != nil {
223 return nil, err
224 }
225 return &containerSandbox{d: d, slot: slot, id: id}, nil
226}
227
228func (d *Driver) release(slot int) {
229 d.mu.Lock()
230 d.slots.Release(slot)
231 d.mu.Unlock()
232}
233
234func (d *Driver) markLeaked(slot int) {
235 d.mu.Lock()
236 d.slots.MarkLeaked(slot)
237 d.mu.Unlock()
238}
239
240func (d *Driver) runDir(id string) string { return filepath.Join(d.cfg.StateDir, "containers", id) }
241
242// sockDir holds only the executor socket: it is the one directory of the job
243// that is bind-mounted (as /run/ermite), so the job never sees the scratch disk
244// file, nor the other host paths of its own sandbox.
245func (d *Driver) sockDir(id string) string { return filepath.Join(d.runDir(id), "sock") }
246
247// scratchDir is where the loop-mounted scratch disk of a job is mounted.
248func (d *Driver) scratchDir(id string) string { return filepath.Join(d.runDir(id), "scratch") }
249
250// workDir is the job workspace, a directory of the scratch disk.
251func (d *Driver) workDir(id string) string { return filepath.Join(d.scratchDir(id), "work") }
252
253// Where the job workspace and its temporary files appear in the container.
254const (
255 containerWorkspace = "/home/runner/work"
256 containerTmp = "/tmp"
257)
258
259type containerSandbox struct {
260 d *Driver
261 slot int
262 id string
263 gate sandbox.BootGate
264
265 mu sync.Mutex
266 started bool
267 stopped bool
268 leakErr error
269 netID string
270 container string
271 mounted bool
272 listener net.Listener
273 conn net.Conn
274 iface string
275 gateway netip.Addr
276 attached bool
277}
278
279// ID implements sandbox.Sandbox.
280func (s *containerSandbox) ID() string { return s.id }
281
282// Start implements sandbox.Sandbox.
283func (s *containerSandbox) Start(ctx context.Context, spec sandbox.JobSpec) (sandbox.Conn, error) {
284 // A Stop from another goroutine aborts the boot through the gate.
285 bootCtx, err := s.gate.Begin(ctx)
286 if err != nil {
287 return nil, fmt.Errorf("container sandbox: %w", err)
288 }
289 defer s.gate.End()
290
291 s.mu.Lock()
292 defer s.mu.Unlock()
293 if s.started || s.stopped {
294 return nil, errors.New("container sandbox: Start called twice or after Stop")
295 }
296 s.started = true
297 d := s.d
298 log := zerowrap.FromCtx(ctx)
299 spec = d.cfg.Resolve(spec)
300 if d.privileged {
301 // At every job start, not just at driver start: the job log and the
302 // operator's logs must say what kind of isolation this job got.
303 log.Warn().Str("job", spec.JobID).Str("container", s.id).Msg(privilegedWarning)
304 }
305 conn, err := s.run(bootCtx, spec)
306 if err != nil {
307 var tail string
308 if s.container != "" {
309 tail, _ = d.eng.LogsTail(context.WithoutCancel(ctx), s.container, 40)
310 }
311 cerr := s.cleanupLocked(context.WithoutCancel(ctx))
312 if tail != "" {
313 err = fmt.Errorf("%w\ncontainer log tail:\n%s", err, tail)
314 }
315 return nil, errors.Join(err, cerr)
316 }
317 log.Info().Str("container", s.container).Int("slot", s.slot).Str("job", spec.JobID).Bool("privileged", d.privileged).Msg("container running")
318 return conn, nil
319}
320
321// run runs the boot stages in order; each records what it created for cleanup.
322func (s *containerSandbox) run(ctx context.Context, spec sandbox.JobSpec) (sandbox.Conn, error) {
323 netw, err := netpolicy.Subnet(s.d.cfg.Network, s.slot)
324 if err != nil {
325 return nil, err
326 }
327 accepted, err := s.runSocket()
328 if err != nil {
329 return nil, err
330 }
331 if err := s.runScratch(ctx, spec); err != nil {
332 return nil, err
333 }
334 labels := map[string]string{sandbox.LabelJob: spec.JobID}
335 if err := s.runNetwork(ctx, netw, labels); err != nil {
336 return nil, err
337 }
338 exited, err := s.runContainer(ctx, spec, netw, labels)
339 if err != nil {
340 return nil, err
341 }
342 return s.waitExecutor(ctx, accepted, exited)
343}
344
345type acceptResult struct {
346 c net.Conn
347 err error
348}
349
350// runSocket creates the per-job directory and listens on the executor socket in
351// it. The directory exists before anything on the Docker side, so a crash
352// leaves a recognisable one.
353func (s *containerSandbox) runSocket() (<-chan acceptResult, error) {
354 d := s.d
355 for _, dir := range []string{d.runDir(s.id), d.sockDir(s.id)} {
356 if err := os.MkdirAll(dir, 0o711); err != nil {
357 return nil, err
358 }
359 if err := os.Chmod(dir, 0o711); err != nil {
360 return nil, err
361 }
362 }
363 dir := d.sockDir(s.id)
364 sock := filepath.Join(dir, sandbox.ContainerSocket)
365 // sockaddr_un.sun_path holds 108 bytes: say so instead of "invalid argument".
366 if len(sock) > maxSockPath {
367 return nil, fmt.Errorf("executor socket path %q is %d bytes, over the %d-byte limit of a Unix socket: use a shorter %s", sock, len(sock), maxSockPath, sandbox.EnvStateDir)
368 }
369 ln, err := net.Listen("unix", sock)
370 if err != nil {
371 return nil, fmt.Errorf("listen executor socket: %w", err)
372 }
373 s.listener = ln
374 // The container's root may be a mapped (sysbox) user: it must be able to
375 // connect. The path is private to the job (random id, 0711 directory).
376 if err := os.Chmod(sock, 0o666); err != nil {
377 return nil, err
378 }
379 accepted := make(chan acceptResult, 1)
380 go func() {
381 c, err := ln.Accept()
382 accepted <- acceptResult{c, err}
383 }()
384 return accepted, nil
385}
386
387// runScratch creates the sparse ext4 scratch disk (ERMITE_RUNNER_JOB_DISK) and
388// loop-mounts it; the workspace directory lives on it.
389func (s *containerSandbox) runScratch(ctx context.Context, spec sandbox.JobSpec) error {
390 d := s.d
391 file := filepath.Join(d.runDir(s.id), "scratch.ext4")
392 if err := disk.CreateScratch(ctx, d.opt.Run, file, spec.DiskMiB, -1); err != nil {
393 return err
394 }
395 mnt := d.scratchDir(s.id)
396 if err := os.MkdirAll(mnt, 0o755); err != nil {
397 return err
398 }
399 if err := d.mnt.Mount(ctx, file, mnt); err != nil {
400 return err
401 }
402 s.mounted = true
403 // Workspace, temp and the nested dockerd's data root on the scratch disk. The
404 // executor (phase G) sets the job layout and ownership inside the sandbox.
405 for _, sub := range []string{"work", "tmp", "docker"} {
406 if err := os.MkdirAll(filepath.Join(mnt, sub), 0o755); err != nil {
407 return err
408 }
409 }
410 return os.Chmod(filepath.Join(mnt, "tmp"), 0o1777)
411}
412
413// runNetwork creates the job's bridge and installs the firewall policy.
414func (s *containerSandbox) runNetwork(ctx context.Context, netw netpolicy.Net, labels map[string]string) error {
415 d := s.d
416 s.iface = fmt.Sprintf("%s%d", bridgePrefix, s.slot)
417 s.gateway = netw.Gateway
418 netID, err := d.eng.CreateNetwork(ctx, networkSpec{Name: "ermite-" + s.id, Bridge: s.iface, Subnet: netw.Prefix, Gateway: netw.Gateway, Labels: labels})
419 if err != nil {
420 return err
421 }
422 s.netID = netID
423 s.attached = true
424 return d.net.Attach(ctx, netpolicy.Guest{Iface: s.iface, Gateway: netw.Gateway, Addr: netw.Guest})
425}
426
427// runContainer creates and starts the container.
428func (s *containerSandbox) runContainer(ctx context.Context, spec sandbox.JobSpec, netw netpolicy.Net, labels map[string]string) (<-chan exitResult, error) {
429 d := s.d
430 entry := sandbox.GuestExecDir + "/" + sandbox.GuestExecName
431 cid, err := d.eng.CreateContainer(ctx, containerSpec{
432 Name: "ermite-" + s.id,
433 Image: d.image,
434 Entrypoint: []string{entry, "init"},
435 Env: []string{"ERMITE_CONNECT=unix:" + sandbox.ContainerRunDir + "/" + sandbox.ContainerSocket},
436 Labels: labels,
437 Runtime: d.runtime,
438 Privileged: d.privileged,
439 Network: "ermite-" + s.id,
440 IPv4: netw.Guest,
441 DNS: netw.Gateway,
442 Binds: []bindMount{
443 {Source: d.exe, Target: entry, ReadOnly: true},
444 {Source: d.sockDir(s.id), Target: sandbox.ContainerRunDir},
445 {Source: d.scratchDir(s.id), Target: sandbox.GuestScratch},
446 {Source: d.workDir(s.id), Target: containerWorkspace},
447 {Source: filepath.Join(d.scratchDir(s.id), "tmp"), Target: containerTmp},
448 },
449 MemoryMiB: spec.MemoryMiB,
450 CPUs: spec.CPUs,
451 Pids: d.cfg.JobPids,
452 })
453 if err != nil {
454 return nil, err
455 }
456 s.container = cid
457 exited := d.eng.WaitExit(ctx, cid)
458 if err := d.eng.StartContainer(ctx, cid); err != nil {
459 return nil, fmt.Errorf("start container: %w", err)
460 }
461 return exited, nil
462}
463
464// waitExecutor waits for the executor to connect. Only ctx (the caller's
465// context, or Stop through the boot gate), the container exiting, and the
466// timeout end the wait.
467func (s *containerSandbox) waitExecutor(ctx context.Context, accepted <-chan acceptResult, exited <-chan exitResult) (sandbox.Conn, error) {
468 timer := time.NewTimer(s.d.opt.ConnectTimeout)
469 defer timer.Stop()
470 select {
471 case r := <-accepted:
472 if r.err != nil {
473 return nil, fmt.Errorf("accept executor connection: %w", r.err)
474 }
475 s.conn = r.c
476 return r.c, nil
477 case ex := <-exited:
478 if ex.Err != nil {
479 return nil, fmt.Errorf("waiting for the container: %w", ex.Err)
480 }
481 return nil, fmt.Errorf("container exited (code %d) before the executor connected", ex.Code)
482 case <-timer.C:
483 return nil, fmt.Errorf("executor did not connect within %s", s.d.opt.ConnectTimeout)
484 case <-ctx.Done():
485 return nil, context.Cause(ctx)
486 }
487}
488
489// Stop implements sandbox.Sandbox.
490func (s *containerSandbox) Stop(ctx context.Context) error {
491 // Abort a boot in progress first: Start holds s.mu for its whole duration.
492 s.gate.Abort()
493 s.mu.Lock()
494 defer s.mu.Unlock()
495 if s.stopped {
496 return s.leakErr
497 }
498 if !s.started {
499 s.releaseLocked()
500 return nil
501 }
502 return s.cleanupLocked(context.WithoutCancel(ctx))
503}
504
505func (s *containerSandbox) releaseLocked() {
506 s.stopped = true
507 s.d.release(s.slot)
508}
509
510// cleanupLocked removes the container first; only when it is confirmed gone are
511// the firewall policy, the network, the scratch mount and the files removed. A
512// container that may still run keeps its policy and bridge: tearing the policy
513// down under it would remove the isolation it relies on. Failures mark the
514// sandbox as leaked (sandbox.ErrLeaked): its slot stays taken until Reap.
515func (s *containerSandbox) cleanupLocked(ctx context.Context) error {
516 d := s.d
517 var errs []error
518 if s.conn != nil {
519 _ = s.conn.Close()
520 s.conn = nil
521 }
522 if s.listener != nil {
523 _ = s.listener.Close()
524 s.listener = nil
525 }
526 if s.container != "" {
527 if err := d.eng.RemoveContainer(ctx, s.container); err != nil {
528 errs = append(errs, fmt.Errorf("remove container: %w", err))
529 } else {
530 s.container = ""
531 }
532 }
533 if s.container == "" {
534 if s.attached {
535 if err := d.net.Detach(ctx, s.iface, s.gateway); err != nil {
536 errs = append(errs, err)
537 } else {
538 s.attached = false
539 }
540 }
541 if s.netID != "" {
542 if err := d.eng.RemoveNetwork(ctx, s.netID); err != nil {
543 errs = append(errs, fmt.Errorf("remove network: %w", err))
544 } else {
545 s.netID = ""
546 }
547 }
548 if s.mounted {
549 if err := d.mnt.Unmount(ctx, d.scratchDir(s.id)); err != nil {
550 errs = append(errs, err)
551 } else {
552 s.mounted = false
553 }
554 }
555 // Never delete files below a mount that is still there.
556 if !s.mounted {
557 if err := os.RemoveAll(d.runDir(s.id)); err != nil {
558 errs = append(errs, err)
559 }
560 }
561 }
562 if len(errs) == 0 {
563 s.releaseLocked()
564 return nil
565 }
566 s.stopped = true
567 d.markLeaked(s.slot)
568 s.leakErr = fmt.Errorf("%w: container sandbox %s: %w", sandbox.ErrLeaked, s.id, errors.Join(errs...))
569 log := zerowrap.FromCtx(ctx)
570 log.Error().Err(s.leakErr).Str("job", s.id).Msg("container cleanup incomplete; call Reap to finish it")
571 return s.leakErr
572}
573
574// Reap implements sandbox.Driver: remove every container and network carrying
575// the job label, the nftables table, the scratch mounts and the per-job
576// directories, and free the slots of leaked sandboxes. It returns ErrBusy while
577// a sandbox is running.
578func (d *Driver) Reap(ctx context.Context) error {
579 log := zerowrap.FromCtx(ctx)
580 d.mu.Lock()
581 busy := d.slots.Busy()
582 d.mu.Unlock()
583 if busy > 0 {
584 return fmt.Errorf("%w (%d running)", sandbox.ErrBusy, busy)
585 }
586 var errs []error
587 ids, err := d.eng.ListContainers(ctx, sandbox.LabelJob)
588 if err != nil {
589 errs = append(errs, fmt.Errorf("list containers: %w", err))
590 }
591 for _, id := range ids {
592 log.Warn().Str("container", id).Msg("removing leftover job container")
593 if err := d.eng.RemoveContainer(ctx, id); err != nil {
594 errs = append(errs, err)
595 }
596 }
597 nets, err := d.eng.ListNetworks(ctx, sandbox.LabelJob)
598 if err != nil {
599 errs = append(errs, fmt.Errorf("list networks: %w", err))
600 }
601 for _, n := range nets {
602 log.Warn().Str("network", n).Msg("removing leftover job network")
603 if err := d.eng.RemoveNetwork(ctx, n); err != nil {
604 errs = append(errs, err)
605 }
606 }
607 if err := d.net.Reset(ctx); err != nil {
608 errs = append(errs, fmt.Errorf("reset nftables table: %w", err))
609 }
610 // Unmount before deleting: a directory with a live mount below it must not
611 // be removed (and EBUSY would be the best case).
612 root := filepath.Join(d.cfg.StateDir, "containers")
613 mounts, merr := d.mnt.MountPoints()
614 if merr != nil {
615 errs = append(errs, fmt.Errorf("read mounts: %w", merr))
616 }
617 stuck := map[string]bool{}
618 for _, mp := range mounts {
619 rel, ok := strings.CutPrefix(mp, root+"/")
620 if !ok {
621 continue
622 }
623 id, _, _ := strings.Cut(rel, "/")
624 if !sandbox.ValidID(id) {
625 continue
626 }
627 log.Warn().Str("mount", mp).Msg("unmounting leftover scratch disk")
628 if err := d.mnt.Unmount(ctx, mp); err != nil {
629 stuck[id] = true
630 errs = append(errs, err)
631 }
632 }
633 ents, _ := os.ReadDir(root)
634 for _, e := range ents {
635 if sandbox.ValidID(e.Name()) && !stuck[e.Name()] {
636 if err := os.RemoveAll(filepath.Join(root, e.Name())); err != nil {
637 errs = append(errs, err)
638 }
639 }
640 }
641 if len(errs) == 0 {
642 d.mu.Lock()
643 d.slots.ReleaseLeaked()
644 d.mu.Unlock()
645 }
646 return errors.Join(errs...)
647}