diff --git a/sandboxd/engine/installca_test.go b/sandboxd/engine/installca_test.go index 82a0ecd..8fa3cb0 100644 --- a/sandboxd/engine/installca_test.go +++ b/sandboxd/engine/installca_test.go @@ -81,6 +81,10 @@ type fakeSilkd struct { readErr map[string]string listCalls int readCalls []string + stat map[string]bool + statMisses map[string]int + statErr map[string]string + statCalls []string } func serveFakeSilkd(t *testing.T, path string) *fakeSilkd { @@ -94,6 +98,9 @@ func serveFakeSilkd(t *testing.T, path string) *fakeSilkd { serial: make(map[string][]byte), readMisses: make(map[string]int), readErr: make(map[string]string), + stat: make(map[string]bool), + statMisses: make(map[string]int), + statErr: make(map[string]string), } go func() { for { @@ -137,6 +144,8 @@ func (f *fakeSilkd) serve(conn net.Conn) { f.handleList(conn, req.Path) case "fs_read": f.handleRead(conn, req.Path) + case "fs_stat": + f.handleStat(conn, req.Path) case "exec": f.handleExec(conn, req.Argv, req.Env) } @@ -230,6 +239,27 @@ func (f *fakeSilkd) handleRead(conn net.Conn, path string) { writeFakeSilkdResponse(conn, &wire.Done{}) } +func (f *fakeSilkd) handleStat(conn net.Conn, path string) { + f.mu.Lock() + f.statCalls = append(f.statCalls, path) + ok := f.stat[path] + reply := f.statErr[path] + if f.statMisses[path] > 0 { + f.statMisses[path]-- + ok = false + } + f.mu.Unlock() + if reply != "" { + writeFakeSilkdResponse(conn, &wire.ErrorResp{Kind: wire.KindInternal, Message: reply}) + return + } + if !ok { + writeFakeSilkdResponse(conn, &wire.ErrorResp{Kind: wire.KindNotFound, Message: path}) + return + } + writeFakeSilkdResponse(conn, &wire.Stat{Info: wire.FileInfo{Kind: wire.FileKindFile}}) +} + func writeFakeSilkdResponse(conn net.Conn, resp wire.Response) { buf, err := wire.EncodeResponse(resp) if err != nil { diff --git a/sandboxd/engine/silkd.go b/sandboxd/engine/silkd.go index 25174ee..c0c521f 100644 --- a/sandboxd/engine/silkd.go +++ b/sandboxd/engine/silkd.go @@ -118,3 +118,28 @@ func (e *Engine) silkdReadFile(ctx context.Context, vsockSocket, path string) ([ } return data, nil } + +// silkdStat answers with a single Stat frame and no terminal Done, so it +// cannot ride silkdStream. +func (e *Engine) silkdStat(ctx context.Context, vsockSocket, path string) error { + s, err := e.dialSilkdSession(ctx, vsockSocket) + if err != nil { + return err + } + defer s.close() + if err = s.send(wire.FsStat{Path: path}); err != nil { + return err + } + frame, err := s.recv() + if err != nil { + return err + } + switch resp := frame.(type) { + case *wire.Stat: + return nil + case *wire.ErrorResp: + return fmt.Errorf("silkd %w", resp) + default: + return fmt.Errorf("unexpected silkd frame %q", frame.RespType()) + } +} diff --git a/sandboxd/engine/volume.go b/sandboxd/engine/volume.go index c85b72f..1255bf0 100644 --- a/sandboxd/engine/volume.go +++ b/sandboxd/engine/volume.go @@ -119,16 +119,24 @@ func (e *Engine) findVolumeDevice(ctx context.Context, vsockSocket, name string) } for _, entry := range entries { serial, err := e.silkdReadFile(ctx, vsockSocket, "/sys/block/"+entry.Name+"/serial") - var respErr *wire.ErrorResp - if errors.As(err, &respErr) && respErr.Kind == wire.KindNotFound { + if isNotFound(err) { continue } if err != nil { return "", false, err } - if strings.TrimSpace(string(serial)) == name { - return "/dev/" + entry.Name, true, nil + if strings.TrimSpace(string(serial)) != name { + continue + } + device := "/dev/" + entry.Name + // The kernel publishes /sys/block before devtmpfs creates the node. + if err := e.silkdStat(ctx, vsockSocket, device); err != nil { + if isNotFound(err) { + return "", false, nil + } + return "", false, err } + return device, true, nil } return "", false, nil } @@ -148,3 +156,8 @@ func (e *Engine) diskAttachArgs(vmName string, spec VolumeSpec) ([]string, error } return append(args, "--directio", directIO), nil } + +func isNotFound(err error) bool { + var respErr *wire.ErrorResp + return errors.As(err, &respErr) && respErr.Kind == wire.KindNotFound +} diff --git a/sandboxd/engine/volume_test.go b/sandboxd/engine/volume_test.go index 72ab3c2..687605e 100644 --- a/sandboxd/engine/volume_test.go +++ b/sandboxd/engine/volume_test.go @@ -164,6 +164,29 @@ func TestMountVolumeWaitsForDelayedSysfsSerial(t *testing.T) { } } +func TestMountVolumeWaitsForDelayedDevNode(t *testing.T) { + path := sockPath(t) + fake := serveFakeSilkd(t, path) + configureVolumeDevices(fake) + fake.mu.Lock() + fake.statMisses["/dev/vdc"] = 3 + fake.mu.Unlock() + if err := New("cocoon", nil, nil, false, "").MountVolume( + t.Context(), path, "imagenet", "/datasets/training", false, + ); err != nil { + t.Fatalf("MountVolume: %v", err) + } + + fake.mu.Lock() + defer fake.mu.Unlock() + if fake.listCalls != 4 { + t.Errorf("fs_list calls = %d, want 4 (3 misses + the successful poll)", fake.listCalls) + } + if got := fake.statCalls[len(fake.statCalls)-1]; got != "/dev/vdc" { + t.Errorf("last fs_stat = %q, want target device", got) + } +} + func TestMountVolumeStopsAtFailedStage(t *testing.T) { for _, tt := range []struct { name string @@ -173,6 +196,7 @@ func TestMountVolumeStopsAtFailedStage(t *testing.T) { }{ {"list", func(f *fakeSilkd) { f.listErr = "list failed" }, "wait for volume device imagenet", 0}, {"read", func(f *fakeSilkd) { f.readErr["/sys/block/vda/serial"] = "read failed" }, "wait for volume device imagenet", 0}, + {"stat", func(f *fakeSilkd) { f.statErr["/dev/vdc"] = "stat failed" }, "wait for volume device imagenet", 0}, {"mkdir", func(f *fakeSilkd) { f.execCode, f.execFailAt = 3, 1 }, "create volume mount point", 1}, {"mount", func(f *fakeSilkd) { f.execCode, f.execFailAt = 3, 2 }, "mount volume imagenet", 2}, } { @@ -215,6 +239,21 @@ func TestMountVolumeDeviceProbeIsBoundedAndCancelable(t *testing.T) { } } +func TestMountVolumeDevNodeNeverAppearsTimesOut(t *testing.T) { + path := sockPath(t) + fake := serveFakeSilkd(t, path) + configureVolumeDevices(fake) + fake.mu.Lock() + delete(fake.stat, "/dev/vdc") + fake.mu.Unlock() + err := New("cocoon", nil, nil, false, "").MountVolume( + t.Context(), path, "imagenet", "/datasets/training", false, + ) + if !errors.Is(err, context.DeadlineExceeded) || !strings.Contains(err.Error(), "wait for volume device imagenet") { + t.Errorf("got %v, want a bounded wait-for-device timeout", err) + } +} + func configureVolumeDevices(f *fakeSilkd) { f.mu.Lock() defer f.mu.Unlock() @@ -225,4 +264,5 @@ func configureVolumeDevices(f *fakeSilkd) { } f.serial["/sys/block/vda/serial"] = []byte("root\n") f.serial["/sys/block/vdc/serial"] = []byte("imagenet\n") + f.stat["/dev/vdc"] = true }