feat(runtime): expose precise SnapshotProgress during setup

api/callback.openapi.yaml

@@ -8,6 +8,28 @@ servers:
 tags:
   - name: worker
 paths:
+  /internal/v1/workers/{runtime_id}/progress:
+    post:
+      operationId: reportWorkerProgress
+      tags: [worker]
+      summary: Report progress for a pending worker action
+      parameters:
+        - $ref: '#/components/parameters/Runtime'
+      requestBody:
+        required: true
+        content:
+          application/json:
+            schema:
+              $ref: '#/components/schemas/WorkerProgress'
+      responses:
+        '204':
+          description: Worker progress accepted
+        '400':
+          description: Invalid worker progress
+        '401':
+          description: Invalid workload identity
+        '409':
+          description: Worker action is no longer pending
   /internal/v1/workers/{runtime_id}/complete:
     post:
       operationId: completeWorker
@@ -37,6 +59,15 @@ components:
       schema:
         type: string
   schemas:
+    WorkerProgress:
+      type: object
+      required: [percentage]
+      properties:
+        percentage:
+          type: integer
+          format: int64
+          minimum: 0
+          maximum: 100
     WorkerResult:
       type: object
       properties:

apps/druid/adapters/cli/callback.go

@@ -12,6 +12,26 @@ type runtimeCallbackHandler struct {
 	allowUnauthenticated bool
 }
 
+func (h runtimeCallbackHandler) ReportWorkerProgress(c *fiber.Ctx, runtimeID callbackapi.Runtime) error {
+	var report callbackapi.WorkerProgress
+	if err := c.BodyParser(&report); err != nil {
+		return fiber.NewError(fiber.StatusBadRequest, "invalid progress report")
+	}
+	if report.Percentage < 0 || report.Percentage > 100 {
+		return fiber.NewError(fiber.StatusBadRequest, "percentage must be between 0 and 100")
+	}
+	if !h.allowUnauthenticated {
+		identity, ok := c.Locals("druid-workload-identity").(ports.RuntimeWorkloadIdentity)
+		if !ok || identity.Kind != "worker" || identity.RuntimeID != string(runtimeID) {
+			return fiber.NewError(fiber.StatusForbidden, "worker identity does not match runtime")
+		}
+	}
+	if err := h.callbacks.ReportProgress(string(runtimeID), report.Percentage); err != nil {
+		return fiber.NewError(fiber.StatusConflict, err.Error())
+	}
+	return c.SendStatus(fiber.StatusNoContent)
+}
+
 func (h runtimeCallbackHandler) CompleteWorker(c *fiber.Ctx, runtimeID callbackapi.Runtime) error {
 	var result callbackapi.WorkerResult
 	if err := c.BodyParser(&result); err != nil {

apps/druid/adapters/cli/callback_test.go

@@ -0,0 +1,40 @@
+package cli
+
+import (
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+
+	"github.com/gofiber/fiber/v2"
+	appservices "github.com/highcard-dev/daemon/apps/druid/core/services"
+	"github.com/highcard-dev/daemon/internal/callbackapi"
+)
+
+func TestRuntimeCallbackHandlerReportsProgress(t *testing.T) {
+	callbacks := appservices.NewWorkerCallbackManager()
+	_, err := callbacks.Register("runtime-1")
+	if err != nil {
+		t.Fatal(err)
+	}
+	handler := runtimeCallbackHandler{callbacks: callbacks, allowUnauthenticated: true}
+	app := fiber.New()
+	callbackapi.RegisterHandlers(app, handler)
+
+	request := httptest.NewRequest(
+		http.MethodPost,
+		"/internal/v1/workers/runtime-1/progress",
+		strings.NewReader(`{"percentage":42}`),
+	)
+	request.Header.Set(fiber.HeaderContentType, fiber.MIMEApplicationJSON)
+	response, err := app.Test(request)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if response.StatusCode != http.StatusNoContent {
+		t.Fatalf("status = %d; want %d", response.StatusCode, http.StatusNoContent)
+	}
+	if progress, ok := callbacks.Progress("runtime-1"); !ok || progress != 42 {
+		t.Fatalf("progress = %v, %v; want 42, true", progress, ok)
+	}
+}

apps/druid/adapters/cli/daemon.go

@@ -190,7 +190,7 @@ func runRuntimeDaemon() error {
 	websocketHandler.SetAllowUnauthenticatedPublic(runtimeAllowUnauthenticatedPublic)
 	handlers := runtimehandlers.RouteHandlers{
 		Server: runtimehandlers.NewRuntimeServer(
-			runtimehandlers.NewHealthHandler(),
+			runtimehandlers.NewHealthHandlerWithProgress(callbacks.Progress),
 			scrollHandler,
 		),
 		Websocket: websocketHandler,

apps/druid/adapters/cli/worker_progress_test.go

@@ -0,0 +1,58 @@
+package cli
+
+import (
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"path/filepath"
+	"testing"
+	"time"
+
+	"github.com/highcard-dev/daemon/internal/core/domain"
+	"github.com/highcard-dev/daemon/internal/core/ports"
+)
+
+func TestWorkerProgressReporterReadsSnapshotProgress(t *testing.T) {
+	reports := make(chan int64, 1)
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, request *http.Request) {
+		var report struct {
+			Percentage int64 `json:"percentage"`
+		}
+		if err := json.NewDecoder(request.Body).Decode(&report); err != nil {
+			t.Error(err)
+		}
+		if got := request.Header.Get("Authorization"); got != "Bearer worker-token" {
+			t.Errorf("authorization = %q; want Bearer worker-token", got)
+		}
+		reports <- report.Percentage
+		w.WriteHeader(http.StatusNoContent)
+	}))
+	defer server.Close()
+
+	progress := domain.NewSnapshotProgress()
+	progress.Percentage.Store(37)
+	tokenFile := filepath.Join(t.TempDir(), "token")
+	if err := os.WriteFile(tokenFile, []byte("worker-token\n"), 0600); err != nil {
+		t.Fatal(err)
+	}
+	stop := startWorkerProgressReporter(
+		ports.RuntimeWorkerAction{
+			RuntimeID:   "runtime-1",
+			CallbackURL: server.URL + "/internal/v1/workers/runtime-1/complete",
+			TokenFile:   tokenFile,
+		},
+		progress,
+		time.Hour,
+	)
+	defer stop()
+
+	select {
+	case percentage := <-reports:
+		if percentage != 37 {
+			t.Fatalf("percentage = %d; want 37", percentage)
+		}
+	case <-time.After(time.Second):
+		t.Fatal("progress was not reported")
+	}
+}

apps/druid/adapters/cli/worker_pull.go

@@ -9,6 +9,7 @@ import (
 	"os"
 	"path/filepath"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/highcard-dev/daemon/internal/callbackapi"
@@ -66,18 +67,22 @@ func runWorkerPull(action ports.RuntimeWorkerAction) ports.RuntimeWorkerResult {
 	if root == "" {
 		root = "/scroll"
 	}
+	progress := domain.NewSnapshotProgress()
+	stopProgress := startWorkerProgressReporter(action, progress, time.Second)
+	defer stopProgress()
+
 	oci := registry.NewOciClient(loadWorkerRegistryStore())
 	digest, err := oci.ResolveDigest(action.Artifact)
 	if err == nil {
 		result.ArtifactDigest = digest
 	}
 	switch action.Mode {
 	case ports.RuntimeWorkerModeUpdate:
-		err = pullWorkerUpdate(root, action.Artifact, oci)
+		err = pullWorkerUpdate(root, action.Artifact, oci, progress)
 	case ports.RuntimeWorkerModeRestore:
-		err = pullWorkerRestore(root, action.Artifact, oci)
+		err = pullWorkerRestore(root, action.Artifact, oci, progress)
 	default:
-		err = pullWorkerCreate(root, action.Artifact, oci)
+		err = pullWorkerCreate(root, action.Artifact, oci, progress)
 	}
 	if err != nil {
 		result.Error = err.Error()
@@ -116,7 +121,7 @@ func loadWorkerRegistryStore() *registry.CredentialStore {
 	return registry.NewCredentialStore(config.Registries)
 }
 
-func pullWorkerCreate(root string, artifact string, oci ports.OciRegistryInterface) error {
+func pullWorkerCreate(root string, artifact string, oci ports.OciRegistryInterface, progress *domain.SnapshotProgress) error {
 	if err := os.MkdirAll(root, 0755); err != nil {
 		return err
 	}
@@ -138,16 +143,16 @@ func pullWorkerCreate(root string, artifact string, oci ports.OciRegistryInterfa
 		}
 		return copyPath(artifact, root)
 	}
-	return oci.PullSelective(root, artifact, true, nil)
+	return oci.PullSelective(root, artifact, true, progress)
 }
 
-func pullWorkerUpdate(root string, artifact string, oci ports.OciRegistryInterface) error {
+func pullWorkerUpdate(root string, artifact string, oci ports.OciRegistryInterface, progress *domain.SnapshotProgress) error {
 	tmp, err := os.MkdirTemp("", "druid-worker-update-*")
 	if err != nil {
 		return err
 	}
 	defer os.RemoveAll(tmp)
-	if err := coreservices.MaterializeScrollArtifact(artifact, tmp, oci, true); err != nil {
+	if err := coreservices.MaterializeScrollArtifactWithProgress(artifact, tmp, oci, true, progress); err != nil {
 		return err
 	}
 	scrollYAML, err := os.ReadFile(filepath.Join(tmp, "scroll.yaml"))
@@ -163,13 +168,13 @@ func pullWorkerUpdate(root string, artifact string, oci ports.OciRegistryInterfa
 	return mergePulledRoot(tmp, root, skipData)
 }
 
-func pullWorkerRestore(root string, artifact string, oci ports.OciRegistryInterface) error {
+func pullWorkerRestore(root string, artifact string, oci ports.OciRegistryInterface, progress *domain.SnapshotProgress) error {
 	tmp, err := os.MkdirTemp("", "druid-worker-restore-*")
 	if err != nil {
 		return err
 	}
 	defer os.RemoveAll(tmp)
-	if err := coreservices.MaterializeScrollArtifact(artifact, tmp, oci, true); err != nil {
+	if err := coreservices.MaterializeScrollArtifactWithProgress(artifact, tmp, oci, true, progress); err != nil {
 		return err
 	}
 	if err := os.MkdirAll(root, 0755); err != nil {
@@ -314,6 +319,69 @@ func copyPath(src string, dst string) error {
 	return err
 }
 
+func startWorkerProgressReporter(action ports.RuntimeWorkerAction, progress *domain.SnapshotProgress, interval time.Duration) func() {
+	if action.CallbackURL == "" || action.TokenFile == "" || progress == nil {
+		return func() {}
+	}
+	suffix := "/internal/v1/workers/" + action.RuntimeID + "/complete"
+	baseURL := strings.TrimSuffix(action.CallbackURL, suffix)
+	if baseURL == action.CallbackURL {
+		return func() {}
+	}
+	client, err := callbackapi.NewClientWithResponses(baseURL)
+	if err != nil {
+		return func() {}
+	}
+	done := make(chan struct{})
+	var wait sync.WaitGroup
+	wait.Add(1)
+	go func() {
+		defer wait.Done()
+		ticker := time.NewTicker(interval)
+		defer ticker.Stop()
+		lastPercentage := int64(-1)
+		report := func() {
+			percentage := progress.Percentage.Load()
+			if percentage == lastPercentage {
+				return
+			}
+			ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
+			response, err := client.ReportWorkerProgressWithResponse(
+				ctx,
+				action.RuntimeID,
+				callbackapi.WorkerProgress{Percentage: percentage},
+				func(_ context.Context, request *http.Request) error {
+					token, err := os.ReadFile(action.TokenFile)
+					if err != nil {
+						return fmt.Errorf("read worker token: %w", err)
+					}
+					request.Header.Set("Authorization", "Bearer "+strings.TrimSpace(string(token)))
+					return nil
+				},
+			)
+			cancel()
+			if err == nil && response.StatusCode() < http.StatusBadRequest {
+				lastPercentage = percentage
+			}
+		}
+		report()
+		for {
+			select {
+			case <-ticker.C:
+				report()
+			case <-done:
+				report()
+				return
+			}
+		}
+	}()
+	var once sync.Once
+	return func() {
+		once.Do(func() { close(done) })
+		wait.Wait()
+	}
+}
+
 func reportWorkerResult(action ports.RuntimeWorkerAction, result ports.RuntimeWorkerResult) error {
 	if action.CallbackURL == "" {
 		body, err := json.Marshal(result)

apps/druid/adapters/cli/worker_test.go

@@ -97,7 +97,7 @@ func TestWorkerRestoreStagesBeforeReplacingRoot(t *testing.T) {
 	mustWrite(t, filepath.Join(root, "data", "old-only.txt"), "old")
 
 	oci := fakeRestoreOCI{t: t}
-	if err := pullWorkerRestore(root, "registry.local/backup:1", oci); err != nil {
+	if err := pullWorkerRestore(root, "registry.local/backup:1", oci, nil); err != nil {
 		t.Fatal(err)
 	}
 

apps/druid/adapters/http/handlers/health_handler.go

@@ -5,12 +5,27 @@ import (
 	"github.com/highcard-dev/daemon/internal/api"
 )
 
-type HealthHandler struct{}
+type ProgressLookup func(runtimeID string) (float64, bool)
+
+type HealthHandler struct {
+	progress ProgressLookup
+}
 
 func NewHealthHandler() *HealthHandler {
 	return &HealthHandler{}
 }
 
+func NewHealthHandlerWithProgress(progress ProgressLookup) *HealthHandler {
+	return &HealthHandler{progress: progress}
+}
+
 func (h *HealthHandler) GetHealthAuth(c *fiber.Ctx) error {
-	return c.JSON(api.HealthResponse{Mode: "ok"})
+	health := api.HealthResponse{Mode: "ok"}
+	if h.progress != nil {
+		if progress, ok := h.progress(c.Params("id")); ok {
+			value := float32(progress)
+			health.Progress = &value
+		}
+	}
+	return c.JSON(health)
 }