feat: backend container executor implement

2026-08-15 20:58:01 +08:00
parent b4e18c6f0c
commit fbce91d797
10 changed files with 1662 additions and 1 deletions
@@ -0,0 +1,361 @@
+// Package backendexecutor prepares and starts one explicitly named backend container.
+// Gateway switching is deliberately outside this package.
+package backendexecutor
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"net/http"
+	"net/url"
+	"os"
+	"path/filepath"
+	"strconv"
+	"strings"
+	"time"
+
+	"github.com/distribution/reference"
+	opencontainersdigest "github.com/opencontainers/go-digest"
+
+	"yms-daemon/internal/containerengine"
+	"yms-daemon/internal/healthcheck"
+	"yms-daemon/internal/transaction"
+)
+
+const (
+	healthPath          = "/yms/actuator/health"
+	healthTimeout       = 120 * time.Second
+	healthInterval      = time.Second
+	hostNetworkMode     = "host"
+	bindMountType       = "bind"
+	stepLoadImage       = "backend.image.load"
+	stepCreateContainer = "backend.container.create"
+	stepStartContainer  = "backend.container.start"
+	stepCheckHealth     = "backend.container.health"
+)
+
+// Request contains exact values supplied by the update package and local deployment configuration.
+// ImageReference is opaque: the executor never extracts meaning from its tag.
+type Request struct {
+	ArchivePath         string
+	ImageReference      string
+	ExpectedImageDigest string
+	Platform            containerengine.Platform
+	ContainerName       string
+	Port                int
+	PortEnvironmentKey  string
+	ConfigSource        string
+	ConfigTarget        string
+	RestartPolicy       containerengine.RestartPolicy
+	HealthEndpoint      string
+}
+
+// Executor drives the persisted transaction up to SWITCHING after the new container is healthy.
+type Executor struct {
+	store       *transaction.Store
+	coordinator *transaction.Coordinator
+	engine      containerengine.Engine
+	checker     *healthcheck.ActuatorChecker
+}
+
+func New(store *transaction.Store, coordinator *transaction.Coordinator, engine containerengine.Engine, httpClient *http.Client) (*Executor, error) {
+	if store == nil {
+		return nil, errors.New("transaction store is required")
+	}
+	if coordinator == nil {
+		return nil, errors.New("transaction coordinator is required")
+	}
+	if engine == nil {
+		return nil, errors.New("container engine is required")
+	}
+	checker, err := healthcheck.NewActuatorChecker(httpClient, healthInterval)
+	if err != nil {
+		return nil, err
+	}
+	return &Executor{store: store, coordinator: coordinator, engine: engine, checker: checker}, nil
+}
+
+// Run resumes from the transaction's persisted state. It does not switch gateway traffic.
+func (e *Executor) Run(ctx context.Context, transactionID string, request Request) error {
+	if strings.TrimSpace(transactionID) == "" {
+		return errors.New("transaction ID is required")
+	}
+	return e.coordinator.RunExclusive(ctx, func(ctx context.Context) error {
+		return e.run(ctx, transactionID, request)
+	})
+}
+
+func (e *Executor) run(ctx context.Context, transactionID string, request Request) error {
+	for {
+		record, err := e.store.Transaction(ctx, transactionID)
+		if err != nil {
+			return err
+		}
+		switch record.State {
+		case transaction.StateCreated:
+			if _, err := e.store.Transition(ctx, transactionID, transaction.StateValidating, "backend container validation started"); err != nil {
+				return err
+			}
+		case transaction.StateValidating:
+			if err := e.validate(ctx, request); err != nil {
+				_, transitionErr := e.store.Transition(ctx, transactionID, transaction.StateFailed, err.Error())
+				return errors.Join(err, transitionErr)
+			}
+			if _, err := e.store.Transition(ctx, transactionID, transaction.StatePrepared, "backend container inputs validated"); err != nil {
+				return err
+			}
+		case transaction.StatePrepared:
+			if err := e.prepare(ctx, transactionID, request); err != nil {
+				return e.failUnlessRecoverable(ctx, transactionID, err)
+			}
+			if _, err := e.store.Transition(ctx, transactionID, transaction.StateStarting, "backend container prepared"); err != nil {
+				return err
+			}
+		case transaction.StateStarting:
+			// 重放 PREPARED 步骤会核对持久化意图，阻止恢复时换入另一组请求参数。
+			if err := e.prepare(ctx, transactionID, request); err != nil {
+				return e.failUnlessRecoverable(ctx, transactionID, err)
+			}
+			if err := e.startAndCheck(ctx, transactionID, request); err != nil {
+				return e.failUnlessRecoverable(ctx, transactionID, err)
+			}
+			if _, err := e.store.Transition(ctx, transactionID, transaction.StateSwitching, "backend container is healthy"); err != nil {
+				return err
+			}
+		case transaction.StateSwitching:
+			return nil
+		default:
+			return fmt.Errorf("backend container executor cannot run transaction %s in state %s", transactionID, record.State)
+		}
+	}
+}
+
+func (e *Executor) failUnlessRecoverable(ctx context.Context, transactionID string, cause error) error {
+	var uncertain *transaction.UncertainStepError
+	if errors.As(cause, &uncertain) || errors.Is(cause, transaction.ErrStepConflict) {
+		return cause
+	}
+	_, transitionErr := e.store.Transition(ctx, transactionID, transaction.StateFailed, cause.Error())
+	return errors.Join(cause, transitionErr)
+}
+
+func (e *Executor) validate(ctx context.Context, request Request) error {
+	if err := validateRequest(request); err != nil {
+		return err
+	}
+	if err := regularFile(request.ArchivePath, "image archive"); err != nil {
+		return err
+	}
+	if err := regularFile(request.ConfigSource, "backend configuration"); err != nil {
+		return err
+	}
+	if err := e.engine.Ping(ctx); err != nil {
+		return err
+	}
+	return nil
+}
+
+func (e *Executor) prepare(ctx context.Context, transactionID string, request Request) error {
+	loadOperation := &loadImageOperation{
+		engine:         e.engine,
+		archivePath:    request.ArchivePath,
+		imageReference: request.ImageReference,
+		expectedDigest: request.ExpectedImageDigest,
+		platform:       request.Platform,
+	}
+	if _, err := e.coordinator.ExecuteStep(ctx, transactionID, loadIntent(request), loadOperation); err != nil {
+		return err
+	}
+
+	image, err := e.engine.InspectImage(ctx, request.ImageReference)
+	if err != nil {
+		return fmt.Errorf("inspect prepared backend image: %w", err)
+	}
+	createOperation := &createContainerOperation{
+		engine:        e.engine,
+		expectedImage: image,
+		spec:          containerSpec(request),
+	}
+	_, err = e.coordinator.ExecuteStep(ctx, transactionID, createIntent(request, image.ID), createOperation)
+	return err
+}
+
+func (e *Executor) startAndCheck(ctx context.Context, transactionID string, request Request) error {
+	startOperation := &startContainerOperation{engine: e.engine, name: request.ContainerName}
+	if _, err := e.coordinator.ExecuteStep(ctx, transactionID, startIntent(request), startOperation); err != nil {
+		return err
+	}
+	healthOperation := &healthOperation{
+		engine:   e.engine,
+		checker:  e.checker,
+		name:     request.ContainerName,
+		endpoint: request.HealthEndpoint,
+		timeout:  healthTimeout,
+	}
+	_, err := e.coordinator.ExecuteStep(ctx, transactionID, healthIntent(request), healthOperation)
+	return err
+}
+
+func validateRequest(request Request) error {
+	if !filepath.IsAbs(request.ArchivePath) {
+		return errors.New("image archive path must be absolute")
+	}
+	if request.ImageReference == "" || strings.TrimSpace(request.ImageReference) != request.ImageReference {
+		return errors.New("exact image reference is required")
+	}
+	if _, err := opencontainersdigest.Parse(request.ExpectedImageDigest); err != nil {
+		return fmt.Errorf("invalid expected image digest: %w", err)
+	}
+	if request.Platform.OS == "" || request.Platform.Architecture == "" {
+		return errors.New("explicit image operating system and architecture are required")
+	}
+	if request.ContainerName == "" || strings.TrimSpace(request.ContainerName) != request.ContainerName {
+		return errors.New("exact container name is required")
+	}
+	if request.Port != 8080 && request.Port != 8081 {
+		return fmt.Errorf("backend container port must be 8080 or 8081: %d", request.Port)
+	}
+	if request.PortEnvironmentKey == "" || strings.Contains(request.PortEnvironmentKey, "=") || strings.TrimSpace(request.PortEnvironmentKey) != request.PortEnvironmentKey {
+		return errors.New("exact port environment key is required")
+	}
+	if !filepath.IsAbs(request.ConfigSource) || !filepath.IsAbs(request.ConfigTarget) {
+		return errors.New("backend configuration source and target must be absolute paths")
+	}
+	if err := validateRestartPolicy(request.RestartPolicy); err != nil {
+		return err
+	}
+	parsed, err := url.ParseRequestURI(request.HealthEndpoint)
+	if err != nil || parsed.Scheme != "http" || parsed.Host == "" || parsed.Path != healthPath {
+		return fmt.Errorf("health endpoint must be an HTTP URL with exact path %s", healthPath)
+	}
+	if parsed.Port() != strconv.Itoa(request.Port) {
+		return fmt.Errorf("health endpoint port must equal backend container port %d", request.Port)
+	}
+	return nil
+}
+
+func validateRestartPolicy(policy containerengine.RestartPolicy) error {
+	switch policy.Name {
+	case "no", "always", "unless-stopped":
+		if policy.MaximumRetryCount != 0 {
+			return fmt.Errorf("restart policy %s does not accept a maximum retry count", policy.Name)
+		}
+	case "on-failure":
+		if policy.MaximumRetryCount < 0 {
+			return errors.New("on-failure maximum retry count cannot be negative")
+		}
+	default:
+		return fmt.Errorf("unsupported explicit restart policy: %q", policy.Name)
+	}
+	return nil
+}
+
+func regularFile(path, description string) error {
+	info, err := os.Stat(path)
+	if err != nil {
+		return fmt.Errorf("inspect %s %s: %w", description, path, err)
+	}
+	if !info.Mode().IsRegular() {
+		return fmt.Errorf("%s is not a regular file: %s", description, path)
+	}
+	return nil
+}
+
+func containerSpec(request Request) containerengine.ContainerSpec {
+	return containerengine.ContainerSpec{
+		Name:           request.ContainerName,
+		ImageReference: request.ImageReference,
+		Platform:       request.Platform,
+		Environment: []string{
+			request.PortEnvironmentKey + "=" + strconv.Itoa(request.Port),
+		},
+		NetworkMode:   hostNetworkMode,
+		RestartPolicy: request.RestartPolicy,
+		Mounts: []containerengine.Mount{{
+			Type:     bindMountType,
+			Source:   request.ConfigSource,
+			Target:   request.ConfigTarget,
+			ReadOnly: true,
+		}},
+	}
+}
+
+func loadIntent(request Request) transaction.StepIntent {
+	return intent(stepLoadImage, "load and verify backend image", struct {
+		ArchivePath    string                   `json:"archivePath"`
+		ImageReference string                   `json:"imageReference"`
+		ImageDigest    string                   `json:"imageDigest"`
+		Platform       containerengine.Platform `json:"platform"`
+	}{request.ArchivePath, request.ImageReference, request.ExpectedImageDigest, request.Platform})
+}
+
+func createIntent(request Request, imageID string) transaction.StepIntent {
+	return intent(stepCreateContainer, "create inactive backend container", struct {
+		Spec    containerengine.ContainerSpec `json:"spec"`
+		ImageID string                        `json:"imageId"`
+	}{containerSpec(request), imageID})
+}
+
+func startIntent(request Request) transaction.StepIntent {
+	return intent(stepStartContainer, "start inactive backend container", struct {
+		ContainerName string `json:"containerName"`
+	}{request.ContainerName})
+}
+
+func healthIntent(request Request) transaction.StepIntent {
+	return intent(stepCheckHealth, "wait for backend Actuator health", struct {
+		ContainerName string        `json:"containerName"`
+		Endpoint      string        `json:"endpoint"`
+		Timeout       time.Duration `json:"timeout"`
+	}{request.ContainerName, request.HealthEndpoint, healthTimeout})
+}
+
+func intent(key, name string, value any) transaction.StepIntent {
+	payload, err := json.Marshal(value)
+	if err != nil {
+		panic(fmt.Sprintf("marshal internal step intent: %v", err))
+	}
+	return transaction.StepIntent{Key: key, Name: name, Intent: payload}
+}
+
+func imageMatches(image containerengine.Image, expectedDigest string, expectedPlatform containerengine.Platform) (bool, error) {
+	expected, err := opencontainersdigest.Parse(expectedDigest)
+	if err != nil {
+		return false, fmt.Errorf("parse expected image digest: %w", err)
+	}
+	if image.Platform != expectedPlatform {
+		return false, nil
+	}
+	if image.DescriptorDigest != "" {
+		actual, err := opencontainersdigest.Parse(image.DescriptorDigest)
+		if err != nil {
+			return false, fmt.Errorf("parse inspected image descriptor digest: %w", err)
+		}
+		if actual == expected {
+			return true, nil
+		}
+	}
+	for _, repoDigest := range image.RepoDigests {
+		parsed, err := reference.ParseAnyReference(repoDigest)
+		if err != nil {
+			return false, fmt.Errorf("parse inspected repository digest %q: %w", repoDigest, err)
+		}
+		digested, ok := parsed.(reference.Digested)
+		if !ok {
+			return false, fmt.Errorf("inspected repository digest is not digest-qualified: %q", repoDigest)
+		}
+		if digested.Digest() == expected {
+			return true, nil
+		}
+	}
+	return false, nil
+}
+
+func resultJSON(value any) json.RawMessage {
+	payload, err := json.Marshal(value)
+	if err != nil {
+		panic(fmt.Sprintf("marshal internal step result: %v", err))
+	}
+	return payload
+}
@@ -0,0 +1,437 @@
+package backendexecutor
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"io"
+	"log/slog"
+	"net/http"
+	"os"
+	"path/filepath"
+	"slices"
+	"sync"
+	"testing"
+
+	"yms-daemon/internal/containerengine"
+	"yms-daemon/internal/transaction"
+)
+
+const (
+	testDigest      = "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
+	testRepository  = "harbor.ymswell.asia/ymswell/glory-ymswell"
+	healthyResponse = `{"status":"UP","components":{"db":{"status":"UP"},"diskSpace":{"status":"UP"},"ping":{"status":"UP"},"redis":{"status":"UP"}}}`
+)
+
+func TestExecutorPreservesOpaqueImageTagsAndReachesSwitching(t *testing.T) {
+	tags := []string{
+		"20260814-093609-d7ed70f0-v1.1.8.1",
+		"20260814-093609-d7ed70f0",
+	}
+	for _, tag := range tags {
+		tag := tag
+		t.Run(tag, func(t *testing.T) {
+			ctx := context.Background()
+			store, coordinator := testTransactionKernel(t)
+			imageReference := testRepository + ":" + tag
+			request := testRequest(t, imageReference)
+			engine := newFakeEngine(request)
+			executor := testExecutor(t, store, coordinator, engine, healthyResponse)
+			record := createTransaction(t, store, "opaque-"+tag)
+
+			if err := executor.Run(ctx, record.ID, request); err != nil {
+				t.Fatalf("run backend executor: %v", err)
+			}
+			current, err := store.Transaction(ctx, record.ID)
+			if err != nil {
+				t.Fatalf("read completed preparation: %v", err)
+			}
+			if current.State != transaction.StateSwitching {
+				t.Fatalf("unexpected transaction state: %s", current.State)
+			}
+			pending, err := store.PendingSteps(ctx, record.ID)
+			if err != nil || len(pending) != 0 {
+				t.Fatalf("unexpected pending steps: steps=%+v err=%v", pending, err)
+			}
+
+			engine.mu.Lock()
+			if engine.lastCreateSpec.ImageReference != imageReference {
+				engine.mu.Unlock()
+				t.Fatalf("image reference changed: got %q want %q", engine.lastCreateSpec.ImageReference, imageReference)
+			}
+			if !slices.Contains(engine.lastCreateSpec.Environment, "SERVER_PORT=8081") {
+				engine.mu.Unlock()
+				t.Fatalf("missing explicit port environment: %+v", engine.lastCreateSpec.Environment)
+			}
+			initialCalls := engine.callCounts()
+			engine.mu.Unlock()
+
+			if err := executor.Run(ctx, record.ID, request); err != nil {
+				t.Fatalf("repeat executor at switching state: %v", err)
+			}
+			engine.mu.Lock()
+			repeatedCalls := engine.callCounts()
+			engine.mu.Unlock()
+			if repeatedCalls != initialCalls {
+				t.Fatalf("switching state repeated engine calls: before=%+v after=%+v", initialCalls, repeatedCalls)
+			}
+		})
+	}
+}
+
+func TestExecutorRecoversRecordedCreateIntentWithoutRepeatingCreate(t *testing.T) {
+	ctx := context.Background()
+	store, coordinator := testTransactionKernel(t)
+	request := testRequest(t, testRepository+":20260814-093609-d7ed70f0-v1.1.8.1")
+	engine := newFakeEngine(request)
+	engine.imageAvailable = true
+	engine.containers[request.ContainerName] = engine.containerFromSpec(containerSpec(request), false)
+	executor := testExecutor(t, store, coordinator, engine, healthyResponse)
+	record := createTransaction(t, store, "recover-create")
+	transitionToPrepared(t, store, record.ID)
+
+	loadStep := loadIntent(request)
+	if _, _, err := store.RecordStepIntent(ctx, record.ID, loadStep); err != nil {
+		t.Fatalf("record completed image intent: %v", err)
+	}
+	if _, err := store.CompleteStep(ctx, record.ID, loadStep.Key, transaction.StepSucceeded, []byte(`{"loaded":true}`), ""); err != nil {
+		t.Fatalf("complete image step: %v", err)
+	}
+	createStep := createIntent(request, engine.loadedImage.ID)
+	if _, _, err := store.RecordStepIntent(ctx, record.ID, createStep); err != nil {
+		t.Fatalf("record crash-window create intent: %v", err)
+	}
+
+	if err := executor.Run(ctx, record.ID, request); err != nil {
+		t.Fatalf("recover backend executor: %v", err)
+	}
+	engine.mu.Lock()
+	calls := engine.callCounts()
+	engine.mu.Unlock()
+	if calls.load != 0 || calls.create != 0 || calls.start != 1 {
+		t.Fatalf("unexpected recovery calls: %+v", calls)
+	}
+	current, err := store.Transaction(ctx, record.ID)
+	if err != nil || current.State != transaction.StateSwitching {
+		t.Fatalf("unexpected recovered state: record=%+v err=%v", current, err)
+	}
+}
+
+func TestExecutorMarksValidationFailureTerminal(t *testing.T) {
+	ctx := context.Background()
+	store, coordinator := testTransactionKernel(t)
+	request := testRequest(t, testRepository+":20260814-093609-d7ed70f0")
+	request.Port = 9090
+	engine := newFakeEngine(request)
+	executor := testExecutor(t, store, coordinator, engine, healthyResponse)
+	record := createTransaction(t, store, "invalid-port")
+
+	if err := executor.Run(ctx, record.ID, request); err == nil {
+		t.Fatal("expected validation failure")
+	}
+	current, err := store.Transaction(ctx, record.ID)
+	if err != nil || current.State != transaction.StateFailed {
+		t.Fatalf("unexpected failed transaction: record=%+v err=%v", current, err)
+	}
+}
+
+func TestExecutorKeepsPreparedStateForConflictingContainerInspection(t *testing.T) {
+	ctx := context.Background()
+	store, coordinator := testTransactionKernel(t)
+	request := testRequest(t, testRepository+":20260814-093609-d7ed70f0")
+	engine := newFakeEngine(request)
+	engine.imageAvailable = true
+	conflicting := engine.containerFromSpec(containerSpec(request), false)
+	conflicting.ImageID = "sha256:bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"
+	engine.containers[request.ContainerName] = conflicting
+	executor := testExecutor(t, store, coordinator, engine, healthyResponse)
+	record := createTransaction(t, store, "container-conflict")
+
+	err := executor.Run(ctx, record.ID, request)
+	var uncertain *transaction.UncertainStepError
+	if !errors.As(err, &uncertain) {
+		t.Fatalf("expected uncertain container step, got %v", err)
+	}
+	current, readErr := store.Transaction(ctx, record.ID)
+	if readErr != nil || current.State != transaction.StatePrepared {
+		t.Fatalf("conflict did not preserve prepared state: record=%+v err=%v", current, readErr)
+	}
+}
+
+func TestExecutorRejectsChangedRecoveryRequestWithoutChangingState(t *testing.T) {
+	ctx := context.Background()
+	store, coordinator := testTransactionKernel(t)
+	request := testRequest(t, testRepository+":20260814-093609-d7ed70f0-v1.1.8.1")
+	engine := newFakeEngine(request)
+	engine.imageAvailable = true
+	executor := testExecutor(t, store, coordinator, engine, healthyResponse)
+	record := createTransaction(t, store, "changed-recovery-request")
+	transitionToPrepared(t, store, record.ID)
+	if err := executor.prepare(ctx, record.ID, request); err != nil {
+		t.Fatalf("prepare original request: %v", err)
+	}
+	if _, err := store.Transition(ctx, record.ID, transaction.StateStarting, "test starting"); err != nil {
+		t.Fatalf("transition to starting: %v", err)
+	}
+
+	changed := request
+	changed.ImageReference = testRepository + ":20260814-093609-d7ed70f0"
+	err := executor.Run(ctx, record.ID, changed)
+	if !errors.Is(err, transaction.ErrStepConflict) {
+		t.Fatalf("expected persisted intent conflict, got %v", err)
+	}
+	current, readErr := store.Transaction(ctx, record.ID)
+	if readErr != nil || current.State != transaction.StateStarting {
+		t.Fatalf("changed recovery request altered transaction: record=%+v err=%v", current, readErr)
+	}
+}
+
+func TestImageMatchesRequiresDigestEvidenceAndExactPlatform(t *testing.T) {
+	t.Parallel()
+	platform := containerengine.Platform{OS: "linux", Architecture: "arm64"}
+	image := containerengine.Image{
+		ID:          "sha256:cccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc",
+		RepoDigests: []string{testRepository + "@" + testDigest},
+		Platform:    platform,
+	}
+	matched, err := imageMatches(image, testDigest, platform)
+	if err != nil || !matched {
+		t.Fatalf("expected exact image match: matched=%t err=%v", matched, err)
+	}
+	image.RepoDigests = nil
+	matched, err = imageMatches(image, testDigest, platform)
+	if err != nil || matched {
+		t.Fatalf("image ID alone must not satisfy manifest digest: matched=%t err=%v", matched, err)
+	}
+	image.DescriptorDigest = testDigest
+	matched, err = imageMatches(image, testDigest, containerengine.Platform{OS: "linux", Architecture: "amd64"})
+	if err != nil || matched {
+		t.Fatalf("platform mismatch accepted: matched=%t err=%v", matched, err)
+	}
+}
+
+func testRequest(t *testing.T, imageReference string) Request {
+	t.Helper()
+	directory := t.TempDir()
+	archivePath := filepath.Join(directory, "backend-image.tar")
+	configPath := filepath.Join(directory, "yms.yaml")
+	if err := os.WriteFile(archivePath, []byte("image archive"), 0o600); err != nil {
+		t.Fatalf("write image archive: %v", err)
+	}
+	if err := os.WriteFile(configPath, []byte("server: {}\n"), 0o600); err != nil {
+		t.Fatalf("write backend configuration: %v", err)
+	}
+	return Request{
+		ArchivePath:         archivePath,
+		ImageReference:      imageReference,
+		ExpectedImageDigest: testDigest,
+		Platform:            containerengine.Platform{OS: "linux", Architecture: "amd64"},
+		ContainerName:       "explicit-backend-8081",
+		Port:                8081,
+		PortEnvironmentKey:  "SERVER_PORT",
+		ConfigSource:        configPath,
+		ConfigTarget:        "/app/config/application.yaml",
+		RestartPolicy:       containerengine.RestartPolicy{Name: "unless-stopped"},
+		HealthEndpoint:      "http://127.0.0.1:8081/yms/actuator/health",
+	}
+}
+
+func testTransactionKernel(t *testing.T) (*transaction.Store, *transaction.Coordinator) {
+	t.Helper()
+	store, err := transaction.OpenStore(context.Background(), filepath.Join(t.TempDir(), "transactions.db"))
+	if err != nil {
+		t.Fatalf("open transaction store: %v", err)
+	}
+	t.Cleanup(func() { _ = store.Close() })
+	coordinator, err := transaction.NewCoordinator(store, slog.New(slog.NewTextHandler(io.Discard, nil)))
+	if err != nil {
+		t.Fatalf("create transaction coordinator: %v", err)
+	}
+	return store, coordinator
+}
+
+func testExecutor(t *testing.T, store *transaction.Store, coordinator *transaction.Coordinator, engine containerengine.Engine, body string) *Executor {
+	t.Helper()
+	client := &http.Client{Transport: roundTripFunc(func(*http.Request) (*http.Response, error) {
+		return &http.Response{StatusCode: http.StatusOK, Header: make(http.Header), Body: io.NopCloser(bytes.NewBufferString(body))}, nil
+	})}
+	executor, err := New(store, coordinator, engine, client)
+	if err != nil {
+		t.Fatalf("create backend executor: %v", err)
+	}
+	return executor
+}
+
+func createTransaction(t *testing.T, store *transaction.Store, suffix string) transaction.Transaction {
+	t.Helper()
+	record, _, err := store.CreateTransaction(context.Background(), transaction.CreateRequest{
+		ID:             "backend-" + suffix,
+		IdempotencyKey: "backend-request-" + suffix,
+		Source:         "test",
+		Service:        "backend",
+	})
+	if err != nil {
+		t.Fatalf("create backend transaction: %v", err)
+	}
+	return record
+}
+
+func transitionToPrepared(t *testing.T, store *transaction.Store, transactionID string) {
+	t.Helper()
+	ctx := context.Background()
+	if _, err := store.Transition(ctx, transactionID, transaction.StateValidating, "test validating"); err != nil {
+		t.Fatalf("transition to validating: %v", err)
+	}
+	if _, err := store.Transition(ctx, transactionID, transaction.StatePrepared, "test prepared"); err != nil {
+		t.Fatalf("transition to prepared: %v", err)
+	}
+}
+
+type roundTripFunc func(*http.Request) (*http.Response, error)
+
+func (f roundTripFunc) RoundTrip(request *http.Request) (*http.Response, error) {
+	return f(request)
+}
+
+type engineCalls struct {
+	ping    int
+	load    int
+	inspect int
+	create  int
+	start   int
+	remove  int
+}
+
+type fakeEngine struct {
+	mu             sync.Mutex
+	request        Request
+	loadedImage    containerengine.Image
+	imageAvailable bool
+	containers     map[string]containerengine.Container
+	lastCreateSpec containerengine.ContainerSpec
+	pingCalls      int
+	loadCalls      int
+	inspectCalls   int
+	createCalls    int
+	startCalls     int
+	removeCalls    int
+}
+
+func newFakeEngine(request Request) *fakeEngine {
+	return &fakeEngine{
+		request: request,
+		loadedImage: containerengine.Image{
+			ID:          "sha256:cccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc",
+			RepoDigests: []string{testRepository + "@" + testDigest},
+			Platform:    request.Platform,
+		},
+		containers: make(map[string]containerengine.Container),
+	}
+}
+
+func (e *fakeEngine) Ping(context.Context) error {
+	e.mu.Lock()
+	defer e.mu.Unlock()
+	e.pingCalls++
+	return nil
+}
+
+func (e *fakeEngine) LoadImage(_ context.Context, input io.Reader) error {
+	e.mu.Lock()
+	defer e.mu.Unlock()
+	e.loadCalls++
+	if _, err := io.ReadAll(input); err != nil {
+		return err
+	}
+	e.imageAvailable = true
+	return nil
+}
+
+func (e *fakeEngine) InspectImage(context.Context, string) (containerengine.Image, error) {
+	e.mu.Lock()
+	defer e.mu.Unlock()
+	e.inspectCalls++
+	if !e.imageAvailable {
+		return containerengine.Image{}, containerengine.ErrNotFound
+	}
+	return e.loadedImage, nil
+}
+
+func (e *fakeEngine) CreateContainer(_ context.Context, spec containerengine.ContainerSpec) (containerengine.Container, error) {
+	e.mu.Lock()
+	defer e.mu.Unlock()
+	e.createCalls++
+	e.lastCreateSpec = spec
+	if _, exists := e.containers[spec.Name]; exists {
+		return containerengine.Container{}, errors.New("container name already exists")
+	}
+	record := e.containerFromSpec(spec, false)
+	e.containers[spec.Name] = record
+	return record, nil
+}
+
+func (e *fakeEngine) StartContainer(_ context.Context, name string) error {
+	e.mu.Lock()
+	defer e.mu.Unlock()
+	e.startCalls++
+	record, exists := e.containers[name]
+	if !exists {
+		return containerengine.ErrNotFound
+	}
+	record.Running = true
+	record.Status = "running"
+	e.containers[name] = record
+	return nil
+}
+
+func (e *fakeEngine) InspectContainer(_ context.Context, name string) (containerengine.Container, error) {
+	e.mu.Lock()
+	defer e.mu.Unlock()
+	record, exists := e.containers[name]
+	if !exists {
+		return containerengine.Container{}, containerengine.ErrNotFound
+	}
+	return record, nil
+}
+
+func (e *fakeEngine) RemoveContainer(_ context.Context, name string, _ bool) error {
+	e.mu.Lock()
+	defer e.mu.Unlock()
+	e.removeCalls++
+	if _, exists := e.containers[name]; !exists {
+		return containerengine.ErrNotFound
+	}
+	delete(e.containers, name)
+	return nil
+}
+
+func (e *fakeEngine) Close() error { return nil }
+
+func (e *fakeEngine) containerFromSpec(spec containerengine.ContainerSpec, running bool) containerengine.Container {
+	return containerengine.Container{
+		ID:             "container-id-" + spec.Name,
+		Name:           spec.Name,
+		ImageID:        e.loadedImage.ID,
+		ImageReference: spec.ImageReference,
+		Platform:       spec.Platform.OS + "/" + spec.Platform.Architecture,
+		Running:        running,
+		Status:         "created",
+		Environment:    append([]string(nil), spec.Environment...),
+		NetworkMode:    spec.NetworkMode,
+		RestartPolicy:  spec.RestartPolicy,
+		Mounts:         append([]containerengine.Mount(nil), spec.Mounts...),
+	}
+}
+
+func (e *fakeEngine) callCounts() engineCalls {
+	return engineCalls{
+		ping:    e.pingCalls,
+		load:    e.loadCalls,
+		inspect: e.inspectCalls,
+		create:  e.createCalls,
+		start:   e.startCalls,
+		remove:  e.removeCalls,
+	}
+}
+
+var _ containerengine.Engine = (*fakeEngine)(nil)
@@ -0,0 +1,200 @@
+package backendexecutor
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"os"
+	"slices"
+	"sync"
+	"time"
+
+	"yms-daemon/internal/containerengine"
+	"yms-daemon/internal/healthcheck"
+	"yms-daemon/internal/transaction"
+)
+
+type loadImageOperation struct {
+	engine         containerengine.Engine
+	archivePath    string
+	imageReference string
+	expectedDigest string
+	platform       containerengine.Platform
+}
+
+func (o *loadImageOperation) Apply(ctx context.Context) error {
+	archive, err := os.Open(o.archivePath)
+	if err != nil {
+		return fmt.Errorf("open image archive: %w", err)
+	}
+	defer archive.Close()
+	return o.engine.LoadImage(ctx, archive)
+}
+
+func (o *loadImageOperation) Inspect(ctx context.Context) (transaction.Inspection, error) {
+	image, err := o.engine.InspectImage(ctx, o.imageReference)
+	if errors.Is(err, containerengine.ErrNotFound) {
+		return transaction.Inspection{Status: transaction.InspectionNotApplied}, nil
+	}
+	if err != nil {
+		return transaction.Inspection{}, err
+	}
+	matches, err := imageMatches(image, o.expectedDigest, o.platform)
+	if err != nil {
+		return transaction.Inspection{}, err
+	}
+	result := resultJSON(struct {
+		ImageID          string                   `json:"imageId"`
+		RepoDigests      []string                 `json:"repoDigests"`
+		DescriptorDigest string                   `json:"descriptorDigest"`
+		Platform         containerengine.Platform `json:"platform"`
+	}{image.ID, image.RepoDigests, image.DescriptorDigest, image.Platform})
+	if !matches {
+		return transaction.Inspection{Status: transaction.InspectionNotApplied, Result: result}, nil
+	}
+	return transaction.Inspection{Status: transaction.InspectionApplied, Result: result}, nil
+}
+
+type createContainerOperation struct {
+	engine        containerengine.Engine
+	expectedImage containerengine.Image
+	spec          containerengine.ContainerSpec
+}
+
+func (o *createContainerOperation) Apply(ctx context.Context) error {
+	_, err := o.engine.CreateContainer(ctx, o.spec)
+	return err
+}
+
+func (o *createContainerOperation) Inspect(ctx context.Context) (transaction.Inspection, error) {
+	record, err := o.engine.InspectContainer(ctx, o.spec.Name)
+	if errors.Is(err, containerengine.ErrNotFound) {
+		return transaction.Inspection{Status: transaction.InspectionNotApplied}, nil
+	}
+	if err != nil {
+		return transaction.Inspection{}, err
+	}
+	result := containerResult(record)
+	if !containerMatches(record, o.expectedImage.ID, o.spec) {
+		return transaction.Inspection{Status: transaction.InspectionUnknown, Result: result}, nil
+	}
+	return transaction.Inspection{Status: transaction.InspectionApplied, Result: result}, nil
+}
+
+type startContainerOperation struct {
+	engine containerengine.Engine
+	name   string
+}
+
+func (o *startContainerOperation) Apply(ctx context.Context) error {
+	return o.engine.StartContainer(ctx, o.name)
+}
+
+func (o *startContainerOperation) Inspect(ctx context.Context) (transaction.Inspection, error) {
+	record, err := o.engine.InspectContainer(ctx, o.name)
+	if errors.Is(err, containerengine.ErrNotFound) {
+		return transaction.Inspection{Status: transaction.InspectionNotApplied}, nil
+	}
+	if err != nil {
+		return transaction.Inspection{}, err
+	}
+	result := containerResult(record)
+	if record.Running && !record.Dead {
+		return transaction.Inspection{Status: transaction.InspectionApplied, Result: result}, nil
+	}
+	return transaction.Inspection{Status: transaction.InspectionNotApplied, Result: result}, nil
+}
+
+type healthOperation struct {
+	engine   containerengine.Engine
+	checker  *healthcheck.ActuatorChecker
+	name     string
+	endpoint string
+	timeout  time.Duration
+
+	mu              sync.Mutex
+	confirmedReport healthcheck.ActuatorReport
+	confirmed       bool
+}
+
+func (o *healthOperation) Apply(ctx context.Context) error {
+	report, err := o.checker.Wait(ctx, o.endpoint, o.timeout, o.running)
+	if err != nil {
+		return err
+	}
+	o.mu.Lock()
+	o.confirmedReport = report
+	o.confirmed = true
+	o.mu.Unlock()
+	return nil
+}
+
+func (o *healthOperation) Inspect(ctx context.Context) (transaction.Inspection, error) {
+	o.mu.Lock()
+	if o.confirmed {
+		report := o.confirmedReport
+		o.mu.Unlock()
+		return healthInspection(report, true, nil)
+	}
+	o.mu.Unlock()
+	report, ready, err := o.checker.Check(ctx, o.endpoint, o.running)
+	return healthInspection(report, ready, err)
+}
+
+func (o *healthOperation) running(ctx context.Context) (bool, error) {
+	record, err := o.engine.InspectContainer(ctx, o.name)
+	if errors.Is(err, containerengine.ErrNotFound) {
+		return false, nil
+	}
+	if err != nil {
+		return false, err
+	}
+	return record.Running && !record.Dead, nil
+}
+
+func healthInspection(report healthcheck.ActuatorReport, ready bool, err error) (transaction.Inspection, error) {
+	result := resultJSON(report)
+	if errors.Is(err, healthcheck.ErrWorkloadStopped) {
+		return transaction.Inspection{Status: transaction.InspectionNotApplied, Result: result}, nil
+	}
+	if err != nil {
+		return transaction.Inspection{Status: transaction.InspectionNotApplied, Result: result}, nil
+	}
+	if !ready {
+		return transaction.Inspection{Status: transaction.InspectionNotApplied, Result: result}, nil
+	}
+	return transaction.Inspection{Status: transaction.InspectionApplied, Result: result}, nil
+}
+
+func containerMatches(record containerengine.Container, expectedImageID string, spec containerengine.ContainerSpec) bool {
+	if record.ImageID != expectedImageID || record.NetworkMode != spec.NetworkMode || record.RestartPolicy != spec.RestartPolicy {
+		return false
+	}
+	for _, expected := range spec.Environment {
+		if !slices.Contains(record.Environment, expected) {
+			return false
+		}
+	}
+	for _, expected := range spec.Mounts {
+		if !slices.Contains(record.Mounts, expected) {
+			return false
+		}
+	}
+	return true
+}
+
+func containerResult(record containerengine.Container) json.RawMessage {
+	return resultJSON(struct {
+		ID      string `json:"id"`
+		ImageID string `json:"imageId"`
+		Running bool   `json:"running"`
+		Dead    bool   `json:"dead"`
+		Status  string `json:"status"`
+	}{record.ID, record.ImageID, record.Running, record.Dead, record.Status})
+}
+
+var _ transaction.Operation = (*loadImageOperation)(nil)
+var _ transaction.Operation = (*createContainerOperation)(nil)
+var _ transaction.Operation = (*startContainerOperation)(nil)
+var _ transaction.Operation = (*healthOperation)(nil)