engine: wait for the device node, not just the sysfs serial

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 {

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())
+	}
+}

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
+}

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
 }