daemon: seed repo volume via CopyToContainer (host-path bind bug → empty workspace); web: spawn poll reads fresh session list
This commit is contained in:
Raphael Westphal
+12
-17
@@ -424,37 +424,32 @@ func (s *Spawner) ensureRepoVolume(ctx context.Context, name string) (bool, erro
|
||||
return true, nil
|
||||
}
|
||||
|
||||
// seedVolume copies the cloned repo into the fresh named volume via a
|
||||
// one-shot container from the worker image (no extra images needed).
|
||||
// seedVolume populates the fresh named volume with the cloned repo by
|
||||
// streaming a tar of the clone into a paused container via the docker API
|
||||
// (CopyToContainer). No bind mounts: bind sources are HOST paths, but the
|
||||
// clone lives inside the daemon container's filesystem.
|
||||
func (s *Spawner) seedVolume(ctx context.Context, slug, repoVolume string) error {
|
||||
repoDir := filepath.Join(s.reposDir, slug)
|
||||
cfg := &container.Config{
|
||||
Image: imageRefWorker,
|
||||
Cmd: []string{"sh", "-c", "cp -a /src/. " + workspaceMount + "/"},
|
||||
Cmd: []string{"true"}, // never started; created only as a volume mount point
|
||||
}
|
||||
hostCfg := &container.HostConfig{
|
||||
Binds: []string{
|
||||
repoVolume + ":" + workspaceMount,
|
||||
repoDir + ":/src:ro",
|
||||
},
|
||||
Binds: []string{repoVolume + ":" + workspaceMount},
|
||||
}
|
||||
created, err := s.cli.ContainerCreate(ctx, cfg, hostCfg, nil, nil, "lvmh-seed-"+slug)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer s.removeContainer(context.Background(), created.ID)
|
||||
if err := s.cli.ContainerStart(ctx, created.ID, container.StartOptions{}); err != nil {
|
||||
return err
|
||||
var buf bytes.Buffer
|
||||
if err := tarDir(&buf, repoDir); err != nil {
|
||||
return fmt.Errorf("tar clone: %w", err)
|
||||
}
|
||||
waitCh, errCh := s.cli.ContainerWait(ctx, created.ID, container.WaitConditionNotRunning)
|
||||
select {
|
||||
case <-waitCh:
|
||||
return nil
|
||||
case err := <-errCh:
|
||||
return err
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
if err := s.cli.CopyToContainer(ctx, created.ID, workspaceMount, &buf, container.CopyToContainerOptions{}); err != nil {
|
||||
return fmt.Errorf("copy into volume: %w", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *Spawner) removeContainer(ctx context.Context, id string) {
|
||||
|
||||
+25
-6
@@ -44,11 +44,12 @@ type recordedCreate struct {
|
||||
type fakeDocker struct {
|
||||
mu sync.Mutex
|
||||
|
||||
calls []dockerCall
|
||||
images int // entries served by GET /images/json
|
||||
volume map[string]bool // existing volumes
|
||||
nextID int
|
||||
create []recordedCreate
|
||||
calls []dockerCall
|
||||
images int // entries served by GET /images/json
|
||||
volume map[string]bool // existing volumes
|
||||
nextID int
|
||||
create []recordedCreate
|
||||
archives []int // sizes of accepted CopyToContainer tar streams
|
||||
|
||||
failBuild bool
|
||||
failBuildHTTP bool
|
||||
@@ -57,6 +58,8 @@ type fakeDocker struct {
|
||||
failStop bool
|
||||
failWait bool
|
||||
failVolumeCreate bool
|
||||
failArchive bool
|
||||
archiveHang bool
|
||||
waitHang bool
|
||||
|
||||
unknown []string
|
||||
@@ -241,6 +244,20 @@ func (f *fakeDocker) ServeHTTP(w http.ResponseWriter, r *http.Request) {
|
||||
return
|
||||
}
|
||||
writeJSONNow(w, http.StatusOK, `{"StatusCode":0,"Error":null}`)
|
||||
case call.Method == http.MethodPut && strings.HasSuffix(call.Path, "/archive"):
|
||||
// CopyToContainer: accept the tar stream and record its size.
|
||||
if f.archiveHang {
|
||||
<-r.Context().Done()
|
||||
return
|
||||
}
|
||||
if f.failArchive {
|
||||
writeJSONNow(w, http.StatusInternalServerError, `{"message":"archive failed"}`)
|
||||
return
|
||||
}
|
||||
f.mu.Lock()
|
||||
f.archives = append(f.archives, len(body))
|
||||
f.mu.Unlock()
|
||||
w.WriteHeader(http.StatusOK)
|
||||
case call.Method == http.MethodDelete && strings.HasPrefix(call.Path, "/containers/"):
|
||||
w.WriteHeader(http.StatusNoContent)
|
||||
default:
|
||||
@@ -273,11 +290,13 @@ func useFakeGit(t *testing.T, mode string) string {
|
||||
" noisy) printf '%s\\n' \"" + strings.Repeat("x", 600) + "\"; exit 1;;\n" +
|
||||
" silent) exit 1;;\n" +
|
||||
"esac\n" +
|
||||
"# clone: create a real worktree so seeding (tarDir) has files to copy\n" +
|
||||
"if [ \"$1\" = clone ]; then d=$(eval \"echo \\${$#}\"); mkdir -p \"$d\" && printf 'fake-repo\n' > \"$d/README.md\"; fi\n" +
|
||||
"exit 0\n"
|
||||
if err := os.WriteFile(filepath.Join(dir, "git"), []byte(script), 0o755); err != nil {
|
||||
t.Fatalf("write fake git: %v", err)
|
||||
}
|
||||
t.Setenv("PATH", dir)
|
||||
t.Setenv("PATH", dir+":"+os.Getenv("PATH"))
|
||||
t.Setenv("FAKE_GIT_MODE", mode)
|
||||
return logPath
|
||||
}
|
||||
|
||||
+21
-10
@@ -13,6 +13,7 @@ import (
|
||||
"net/http/httptest"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"reflect"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
@@ -89,16 +90,20 @@ func TestSpawnerStartHappyPath(t *testing.T) {
|
||||
t.Fatalf("network = %q", c.HostConfig.NetworkMode)
|
||||
}
|
||||
|
||||
// fresh repo volume seeded from the clone via a one-shot container
|
||||
// fresh repo volume seeded from the clone via CopyToContainer (tar)
|
||||
seed := f.createsByName("lvmh-seed-")
|
||||
if len(seed) != 1 {
|
||||
t.Fatalf("seed containers = %+v", seed)
|
||||
}
|
||||
if seed[0].Image != imageRefWorker || len(seed[0].Cmd) == 0 || !strings.Contains(seed[0].Cmd[len(seed[0].Cmd)-1], workspaceMount) {
|
||||
if seed[0].Image != imageRefWorker {
|
||||
t.Fatalf("seed create = %+v", seed[0])
|
||||
}
|
||||
if !strings.HasPrefix(seed[0].HostConfig.Binds[1], sp.reposDir) {
|
||||
t.Fatalf("seed binds = %v, want clone dir mounted at /src", seed[0].HostConfig.Binds)
|
||||
wantSeedBinds := []string{"lvmh-repo-group-project:" + workspaceMount}
|
||||
if !reflect.DeepEqual(seed[0].HostConfig.Binds, wantSeedBinds) {
|
||||
t.Fatalf("seed binds = %v, want %v (no host-path binds)", seed[0].HostConfig.Binds, wantSeedBinds)
|
||||
}
|
||||
if len(f.archives) != 1 || f.archives[0] == 0 {
|
||||
t.Fatalf("CopyToContainer archives = %v, want one non-empty tar", f.archives)
|
||||
}
|
||||
|
||||
// no build expected (fake daemon already has the image)
|
||||
@@ -204,11 +209,11 @@ func TestSpawnerRunJobErrorStates(t *testing.T) {
|
||||
wantMsg: "volume create failed",
|
||||
},
|
||||
{
|
||||
name: "seed-wait-fails",
|
||||
name: "seed-copy-fails",
|
||||
setup: func(t *testing.T, f *fakeDocker) {
|
||||
f.failWait = true
|
||||
f.failArchive = true
|
||||
},
|
||||
wantMsg: "wait failed",
|
||||
wantMsg: "copy into volume",
|
||||
},
|
||||
}
|
||||
for _, tc := range cases {
|
||||
@@ -315,17 +320,23 @@ func TestSpawnerRemoveSession(t *testing.T) {
|
||||
f.assertNoUnknown(t)
|
||||
}
|
||||
|
||||
func TestSpawnerSeedVolumeWaitContextCancel(t *testing.T) {
|
||||
func TestSpawnerSeedVolumeCancel(t *testing.T) {
|
||||
useFakeGit(t, fakeGitModeOK)
|
||||
f := newFakeDocker()
|
||||
f.waitHang = true // server never answers wait
|
||||
f.archiveHang = true // server never answers the archive PUT
|
||||
sp, _ := newTestSpawner(t, f)
|
||||
if err := os.MkdirAll(filepath.Join(sp.reposDir, "group-project"), 0o755); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := os.WriteFile(filepath.Join(sp.reposDir, "group-project", "README.md"), []byte("x"), 0o644); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
done := make(chan error, 1)
|
||||
go func() { done <- sp.seedVolume(ctx, "group-project", "lvmh-repo-group-project") }()
|
||||
waitFor(t, 5*time.Second, func() bool { return f.hasCallSuffix(http.MethodPost, "/wait") })
|
||||
waitFor(t, 5*time.Second, func() bool { return f.hasCallSuffix(http.MethodPut, "/archive") })
|
||||
cancel()
|
||||
select {
|
||||
case err := <-done:
|
||||
|
||||
Reference in New Issue
Block a user