fix(runtime): refresh registry-tagged container images

README.md

@@ -69,6 +69,8 @@ Runtime selection is daemon-only: start the daemon with `druid daemon --runtime
 
 Kubernetes runtime support is available with `druid daemon --runtime kubernetes` for in-cluster daemons or out-of-cluster daemons using kubeconfig. It stores daemon scroll state in ConfigMaps, materializes OCI artifacts through `druid worker pull` Jobs, and uses kubelet pod stats for procedure-level traffic checks. See `docs/kubernetes_runtime.md` for kubeconfig, RBAC, and PVC setup.
 
+Both backends refresh registry-tagged container images at startup, including `stable`, `latest`, and version tags. Kubernetes uses `imagePullPolicy: Always`; Docker pulls before creating the container. Existing layers are reused when the registry digest is unchanged. Digest-pinned images and explicit `:local` development images use the cache when available. A tag update takes effect on the next container start; running containers are not automatically restarted.
+
 ## Documentation
 
 Read more at https://docs.druid.gg/cli

internal/runtime/docker/images.go

@@ -2,23 +2,40 @@ package docker
 
 import (
 	"context"
+	"encoding/json"
+	"fmt"
 	"io"
 
 	cerrdefs "github.com/containerd/errdefs"
 	"github.com/docker/docker/api/types/image"
+	"github.com/highcard-dev/daemon/internal/runtime/imagepolicy"
 )
 
 func (b *Backend) pullImage(ctx context.Context, imageRef string) error {
-	if _, err := b.client.ImageInspect(ctx, imageRef); err == nil {
-		return nil
-	} else if !cerrdefs.IsNotFound(err) {
-		return err
+	if !imagepolicy.ShouldRefresh(imageRef) {
+		if _, err := b.client.ImageInspect(ctx, imageRef); err == nil {
+			return nil
+		} else if !cerrdefs.IsNotFound(err) {
+			return err
+		}
 	}
 	reader, err := b.client.ImagePull(ctx, imageRef, image.PullOptions{})
 	if err != nil {
 		return err
 	}
 	defer reader.Close()
-	_, _ = io.Copy(io.Discard, reader)
-	return nil
+	decoder := json.NewDecoder(reader)
+	for {
+		var message struct {
+			Error string `json:"error"`
+		}
+		if err := decoder.Decode(&message); err == io.EOF {
+			return nil
+		} else if err != nil {
+			return err
+		}
+		if message.Error != "" {
+			return fmt.Errorf("pull image %s: %s", imageRef, message.Error)
+		}
+	}
 }

internal/runtime/docker/images_test.go

@@ -0,0 +1,81 @@
+package docker
+
+import (
+	"context"
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+
+	"github.com/docker/docker/client"
+)
+
+func TestPullImageRefreshesRegistryTags(t *testing.T) {
+	digestImage := "druid@sha256:" + strings.Repeat("a", 64)
+	for _, tc := range []struct {
+		name       string
+		image      string
+		cached     bool
+		pullBody   string
+		pullStatus int
+		wantPulls  int
+		wantError  string
+	}{
+		{name: "stable refreshes cached image", image: "registry.example/druid:stable", cached: true, wantPulls: 1},
+		{name: "version tags also refresh", image: "alpine:3.20", cached: true, wantPulls: 1},
+		{name: "digest reuses cache", image: digestImage, cached: true},
+		{name: "local image reuses cache", image: "druid:local", cached: true},
+		{name: "missing digest pulls", image: digestImage, wantPulls: 1},
+		{name: "registry failure does not use stale cache", image: "druid:stable", cached: true, pullStatus: http.StatusInternalServerError, pullBody: `{"message":"registry unavailable"}`, wantPulls: 1, wantError: "registry unavailable"},
+		{name: "stream failure does not use stale cache", image: "druid:stable", cached: true, pullBody: `{"error":"access denied"}`, wantPulls: 1, wantError: "access denied"},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			pulls := 0
+			server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				w.Header().Set("Content-Type", "application/json")
+				switch {
+				case r.Method == http.MethodGet && strings.HasSuffix(r.URL.Path, "/json"):
+					if !tc.cached {
+						w.WriteHeader(http.StatusNotFound)
+						fmt.Fprint(w, `{"message":"image not found"}`)
+						return
+					}
+					fmt.Fprint(w, `{"Id":"sha256:cached"}`)
+				case r.Method == http.MethodPost && strings.HasSuffix(r.URL.Path, "/images/create"):
+					pulls++
+					status := tc.pullStatus
+					if status == 0 {
+						status = http.StatusOK
+					}
+					w.WriteHeader(status)
+					if tc.pullBody != "" {
+						fmt.Fprint(w, tc.pullBody)
+					} else {
+						fmt.Fprint(w, `{"status":"downloaded current image"}`)
+					}
+				default:
+					t.Errorf("unexpected Docker request: %s %s", r.Method, r.URL.Path)
+					w.WriteHeader(http.StatusNotFound)
+				}
+			}))
+			defer server.Close()
+			cli, err := client.NewClientWithOpts(client.WithHost(server.URL), client.WithVersion("1.47"))
+			if err != nil {
+				t.Fatal(err)
+			}
+			defer cli.Close()
+			backend := &Backend{client: cli}
+			err = backend.pullImage(context.Background(), tc.image)
+			if tc.wantError == "" && err != nil {
+				t.Fatal(err)
+			}
+			if tc.wantError != "" && (err == nil || !strings.Contains(err.Error(), tc.wantError)) {
+				t.Fatalf("error = %v, want %q", err, tc.wantError)
+			}
+			if pulls != tc.wantPulls {
+				t.Fatalf("pull requests = %d, want %d", pulls, tc.wantPulls)
+			}
+		})
+	}
+}

internal/runtime/imagepolicy/policy.go

@@ -0,0 +1,9 @@
+package imagepolicy
+
+import "strings"
+
+// ShouldRefresh checks registry tags on each container start. Only digests are
+// immutable; :local images are explicitly built and imported for development.
+func ShouldRefresh(image string) bool {
+	return !strings.Contains(image, "@") && !strings.HasSuffix(image, ":local")
+}

internal/runtime/imagepolicy/policy_test.go

@@ -0,0 +1,27 @@
+package imagepolicy
+
+import "testing"
+
+func TestShouldRefresh(t *testing.T) {
+	for _, tc := range []struct {
+		image string
+		want  bool
+	}{
+		{"artifacts.druid.gg/druid-team/druid-dev:stable", true},
+		{"druid:latest", true},
+		{"alpine", true},
+		{"alpine:3.20", true},
+		{"registry.example:5000/team/image:release", true},
+		{"registry.example:5000/team/image", true},
+		{"druid@sha256:abc", false},
+		{"druid:stable@sha256:abc", false},
+		{"druid:local", false},
+		{"registry.example:5000/druid:local", false},
+	} {
+		t.Run(tc.image, func(t *testing.T) {
+			if got := ShouldRefresh(tc.image); got != tc.want {
+				t.Fatalf("ShouldRefresh(%q) = %v, want %v", tc.image, got, tc.want)
+			}
+		})
+	}
+}

internal/runtime/kubernetes/image_pull_policy_test.go

@@ -0,0 +1,49 @@
+package kubernetes
+
+import (
+	"testing"
+
+	"github.com/highcard-dev/daemon/internal/core/domain"
+	corev1 "k8s.io/api/core/v1"
+)
+
+func TestRuntimeWorkloadsUseImagePullPolicy(t *testing.T) {
+	for _, tc := range []struct {
+		image string
+		want  corev1.PullPolicy
+	}{
+		{"artifacts.druid.gg/druid-team/druid-dev:stable", corev1.PullAlways},
+		{"alpine:3.20", corev1.PullAlways},
+		{"druid@sha256:abc", corev1.PullIfNotPresent},
+		{"druid:local", corev1.PullIfNotPresent},
+	} {
+		t.Run(tc.image, func(t *testing.T) {
+			root := ref("games", "runtime-data")
+			procedure := &domain.Procedure{Image: tc.image}
+			job, err := procedureJobSpec("games", root, "build", "build", "build-0", 0, procedure, nil, "", "")
+			if err != nil {
+				t.Fatal(err)
+			}
+			initName, mainName := "init", "files"
+			command := &domain.CommandInstructionSet{Procedures: []*domain.Procedure{
+				{Id: &initName, Image: tc.image},
+				{Id: &mainName, Image: tc.image},
+			}}
+			sts, err := persistentStatefulSetSpec("games", root, "dev", "dev", command, nil, "")
+			if err != nil {
+				t.Fatal(err)
+			}
+			helper := helperJobSpec("games", "helper", "runtime-data", tc.image, nil, "", nil)
+			for _, container := range []corev1.Container{
+				job.Spec.Template.Spec.Containers[0],
+				sts.Spec.Template.Spec.Containers[0],
+				sts.Spec.Template.Spec.InitContainers[0],
+				helper.Spec.Template.Spec.Containers[0],
+			} {
+				if container.ImagePullPolicy != tc.want {
+					t.Errorf("container %s (%s) pull policy = %s, want %s", container.Name, container.Image, container.ImagePullPolicy, tc.want)
+				}
+			}
+		})
+	}
+}

internal/runtime/kubernetes/resources.go

@@ -14,8 +14,16 @@ import (
 
 	"github.com/highcard-dev/daemon/internal/core/domain"
 	"github.com/highcard-dev/daemon/internal/core/ports"
+	"github.com/highcard-dev/daemon/internal/runtime/imagepolicy"
 )
 
+func imagePullPolicy(image string) corev1.PullPolicy {
+	if imagepolicy.ShouldRefresh(image) {
+		return corev1.PullAlways
+	}
+	return corev1.PullIfNotPresent
+}
+
 func pvcSpec(namespace string, name string, storageClass string, storageRequest string) *corev1.PersistentVolumeClaim {
 	if storageRequest == "" {
 		storageRequest = "1Gi"
@@ -162,10 +170,11 @@ func helperJobSpec(namespace string, jobName string, pvc string, image string, c
 	podSpec := corev1.PodSpec{
 		RestartPolicy: corev1.RestartPolicyNever,
 		Containers: []corev1.Container{{
-			Name:         "main",
-			Image:        image,
-			Command:      command,
-			VolumeMounts: []corev1.VolumeMount{{Name: "scroll", MountPath: "/scroll"}},
+			Name:            "main",
+			Image:           image,
+			ImagePullPolicy: imagePullPolicy(image),
+			Command:         command,
+			VolumeMounts:    []corev1.VolumeMount{{Name: "scroll", MountPath: "/scroll"}},
 		}},
 		Volumes: []corev1.Volume{pvcVolume("scroll", pvc)},
 	}
@@ -209,7 +218,7 @@ func procedureJobSpec(namespace string, root string, commandName string, procedu
 		WorkingDir:      procedure.WorkingDir,
 		TTY:             procedure.TTY,
 		Stdin:           true,
-		ImagePullPolicy: corev1.PullIfNotPresent,
+		ImagePullPolicy: imagePullPolicy(procedure.Image),
 		Env:             envVars(env),
 		VolumeMounts:    volumeMounts(procedure.Mounts),
 	}
@@ -292,7 +301,7 @@ func persistentStatefulSetSpec(namespace string, root string, commandName string
 		WorkingDir:      mainProcedure.WorkingDir,
 		TTY:             mainProcedure.TTY,
 		Stdin:           true,
-		ImagePullPolicy: corev1.PullIfNotPresent,
+		ImagePullPolicy: imagePullPolicy(mainProcedure.Image),
 		Env:             envVars(mainEnv),
 		VolumeMounts:    volumeMounts(mainProcedure.Mounts),
 	}
@@ -316,7 +325,7 @@ func persistentStatefulSetSpec(namespace string, root string, commandName string
 			WorkingDir:      procedure.WorkingDir,
 			TTY:             procedure.TTY,
 			Stdin:           true,
-			ImagePullPolicy: corev1.PullIfNotPresent,
+			ImagePullPolicy: imagePullPolicy(procedure.Image),
 			Env:             envVars(env),
 			VolumeMounts:    volumeMounts(procedure.Mounts),
 		})

test/integration/docker/docker_cli_test.go

@@ -23,7 +23,7 @@ func TestDockerBackendCLIComplexLifecycle(t *testing.T) {
 	callbackPort := e2e.FreePort(t)
 	name := fmt.Sprintf("docker-cli-%d", time.Now().UnixNano())
 	fixture := e2e.WriteFixture(t, filepath.Join(t.TempDir(), "scroll"), name, port, routePort)
-	workerImage := e2e.BuildDockerImage(t, "druid-cli-e2e:"+name)
+	workerImage := e2e.BuildDockerImage(t, "druid-cli-e2e-"+name+":local")
 	containerHost := e2e.DockerHostAddress(t)
 
 	socket := filepath.Join(os.TempDir(), fmt.Sprintf("druid-docker-%d.sock", time.Now().UnixNano()))
@@ -94,7 +94,7 @@ func TestDockerBackendVolumeStorageWorkerLifecycleBackupRestore(t *testing.T) {
 	containerHost := e2e.DockerHostAddress(t)
 	name := fmt.Sprintf("docker-volume-%d", time.Now().UnixNano())
 	fixture := e2e.WriteFixture(t, filepath.Join(t.TempDir(), "scroll"), name, port, routePort)
-	workerImage := e2e.BuildDockerImage(t, "druid-cli-e2e:"+name)
+	workerImage := e2e.BuildDockerImage(t, "druid-cli-e2e-"+name+":local")
 
 	pushArtifact := fmt.Sprintf("127.0.0.1:%d/druid-e2e/%s:v1", registryPort, name)
 	runtimeArtifact := fmt.Sprintf("%s:%d/druid-e2e/%s:v1", containerHost, registryPort, name)
@@ -166,7 +166,7 @@ func TestDockerBackendColdstarterFrontsRuntime(t *testing.T) {
 	publicPort := e2e.FreePort(t)
 	callbackPort := e2e.FreePort(t)
 	name := fmt.Sprintf("docker-coldstart-%d", time.Now().UnixNano())
-	image := e2e.BuildDockerImage(t, "druid-coldstart-e2e:"+name)
+	image := e2e.BuildDockerImage(t, "druid-coldstart-e2e-"+name+":local")
 	fixtureDir := writeColdstarterFixture(t, filepath.Join(t.TempDir(), "scroll"), name, image, runtimePort)
 	containerHost := e2e.DockerHostAddress(t)
 

test/integration/kubernetes/kubernetes_cli_test.go

@@ -29,7 +29,7 @@ func TestKubernetesBackendCLIComplexLifecycle(t *testing.T) {
 	namespace := "druid-cli-e2e-" + suffix
 	name := "k8s-cli-" + suffix
 	fixture := e2e.WriteFixture(t, filepath.Join(t.TempDir(), "scroll"), name, port, routePort)
-	workerImage := e2e.BuildDockerImage(t, "druid-cli-e2e:"+name)
+	workerImage := e2e.BuildDockerImage(t, "druid-cli-e2e-"+name+":local")
 	importImageIntoCurrentCluster(t, workerImage)
 	pushArtifact := fmt.Sprintf("127.0.0.1:%d/druid-e2e/%s:v1", registryPort, name)
 	runtimeArtifact := fmt.Sprintf("%s:%d/druid-e2e/%s:v1", containerHost, registryPort, name)