| ${esc(volume.logical_id)} |
${badge(volume.type)} |
${esc(volume.mount_path)} |
${nodeLink(volume.bound_node)} |
${display(volume.shared_backend_id)} |
+ ${esc(formatBytes(volume.requested_size_bytes || volume.desired_size_bytes))} |
${esc(formatBytes(volume.desired_size_bytes))} |
${esc(formatBytes(volume.applied_size_bytes))} |
- ${badge(volume.state)} |
+ ${badge(volume.state)}${volume.rejected ? ` ${badge(volume.rejected_reason || 'rejected')}` : ''} |
${display(volume.last_error)} |
`);
@@ -357,7 +362,7 @@ function renderServiceDetail(service) {
${section('Revisions', revisions)}
${(service.port_forwards || []).length ? section('Port forwards', table(['Host port', 'VM port'], portRows, '')) : ''}
- ${(service.volumes || []).length ? section('Persistent volumes', table(['Volume', 'Type', 'Mount path', 'Bound node', 'Backend', 'Desired', 'Applied', 'State', 'Last error'], volumeRows, '')) : ''}`;
+ ${(service.volumes || []).length ? section('Persistent volumes', table(['Volume', 'Type', 'Mount path', 'Bound node', 'Backend', 'Requested', 'Effective', 'Applied', 'State', 'Last error'], volumeRows, '')) : ''}`;
}
async function detail(kind, id) {
diff --git a/internal/reconciler/errors.go b/internal/reconciler/errors.go
index 9e23377..0f78984 100644
--- a/internal/reconciler/errors.go
+++ b/internal/reconciler/errors.go
@@ -1,6 +1,10 @@
package reconciler
-import "fmt"
+import (
+ "fmt"
+
+ "github.com/artemnikitin/firework/internal/vm"
+)
// FailureStage identifies the blocking host stage that prevented convergence.
// It is intentionally a small, stable set so agents can publish bounded
@@ -53,3 +57,51 @@ func HasFailureStage(err error, stage FailureStage) bool {
}
return false
}
+
+// IsIncomplete reports whether a reconciliation error consists *only* of benign
+// start-barrier races — a start that was aborted by a concurrent stop or
+// remove, or one that collided with another start still preparing volumes.
+//
+// The distinction matters because of what an ordinary nil return would do. The
+// agent advances lastRevision at the end of a successful tick, and the next
+// tick then takes the unchanged-revision shortcut and never re-plans, leaving
+// an aborted service down until the revision itself changes. So an aborted
+// start must neither succeed nor be reported as a hard failure: it is
+// incomplete, and the caller retries on the next tick without claiming the
+// revision or raising a reconcile_failed condition.
+//
+// A batch that mixes an abort with a genuine failure is a failure. Both
+// classifications leave the revision unchanged; the difference is what the node
+// reports.
+func IsIncomplete(err error) bool {
+ leaves := reconcileLeaves(err, nil)
+ if len(leaves) == 0 {
+ return false
+ }
+ for _, leaf := range leaves {
+ if !vm.IsStartRace(leaf) {
+ return false
+ }
+ }
+ return true
+}
+
+// reconcileLeaves flattens an aggregate error into the individual errors it was
+// built from. Joined branches are walked; a plain wrapped chain is followed to
+// its innermost error, which is where a sentinel lives.
+func reconcileLeaves(err error, out []error) []error {
+ for err != nil {
+ if joined, ok := err.(interface{ Unwrap() []error }); ok {
+ for _, child := range joined.Unwrap() {
+ out = reconcileLeaves(child, out)
+ }
+ return out
+ }
+ wrapped, ok := err.(interface{ Unwrap() error })
+ if !ok || wrapped.Unwrap() == nil {
+ return append(out, err)
+ }
+ err = wrapped.Unwrap()
+ }
+ return out
+}
diff --git a/internal/reconciler/errors_test.go b/internal/reconciler/errors_test.go
index 6eb572c..4a2ebd4 100644
--- a/internal/reconciler/errors_test.go
+++ b/internal/reconciler/errors_test.go
@@ -4,6 +4,8 @@ import (
"errors"
"fmt"
"testing"
+
+ "github.com/artemnikitin/firework/internal/vm"
)
func TestHasFailureStageFindsWrappedAndJoinedStages(t *testing.T) {
@@ -18,3 +20,37 @@ func TestHasFailureStageFindsWrappedAndJoinedStages(t *testing.T) {
t.Fatal("VM stage was lost through aggregate error")
}
}
+
+// The abort must survive the exact wrapping the apply path performs. A
+// hand-built error proves nothing: the defect this guards against is a link in
+// that chain flattening the error with %v.
+func TestIsIncompleteSeesThroughTheProductionErrorShape(t *testing.T) {
+ abort := stageError(FailureStageVM,
+ fmt.Errorf("starting VM: %w", fmt.Errorf("service app: %w", vm.ErrStartAborted)))
+ inProgress := stageError(FailureStageVM,
+ fmt.Errorf("starting VM: %w", fmt.Errorf("service api is in state starting: %w", vm.ErrStartInProgress)))
+ genuine := stageError(FailureStageNetwork, errors.New("tap creation failed"))
+
+ wrapApply := func(errs ...error) error {
+ return combineErrors([]error{fmt.Errorf("reconciliation had %d error(s): %w", len(errs), errors.Join(errs...))})
+ }
+
+ tests := []struct {
+ name string
+ err error
+ want bool
+ }{
+ {name: "nil is not incomplete", err: nil, want: false},
+ {name: "single abort", err: wrapApply(abort), want: true},
+ {name: "abort plus concurrent start", err: wrapApply(abort, inProgress), want: true},
+ {name: "abort mixed with a genuine failure", err: wrapApply(abort, genuine), want: false},
+ {name: "genuine failure alone", err: wrapApply(genuine), want: false},
+ }
+ for _, test := range tests {
+ t.Run(test.name, func(t *testing.T) {
+ if got := IsIncomplete(test.err); got != test.want {
+ t.Fatalf("IsIncomplete = %v, want %v (err: %v)", got, test.want, test.err)
+ }
+ })
+ }
+}
diff --git a/internal/reconciler/reconciler.go b/internal/reconciler/reconciler.go
index 1efc87f..8bf1e03 100644
--- a/internal/reconciler/reconciler.go
+++ b/internal/reconciler/reconciler.go
@@ -18,6 +18,7 @@ import (
"github.com/artemnikitin/firework/internal/healthcheck"
"github.com/artemnikitin/firework/internal/network"
"github.com/artemnikitin/firework/internal/vm"
+ "github.com/artemnikitin/firework/internal/volume"
)
// Action represents a reconciliation action the agent needs to take.
@@ -44,7 +45,7 @@ type VMManager interface {
}
type volumePreflighter interface {
- Preflight(context.Context, config.ServiceConfig) error
+ Preflight(context.Context, config.ServiceConfig) ([]volume.Rejection, error)
}
type vmRecoverer interface {
@@ -552,11 +553,15 @@ func (r *Reconciler) applyAllAtOnce(ctx context.Context, actions []Action) error
case ActionUpdate:
r.logger.Info("updating service (stop + start)", "service", action.Service.Name)
- if err := r.preflight(ctx, action.Service); err != nil {
+ rejections, err := r.preflight(ctx, action.Service)
+ if err != nil {
r.logger.Error("volume preflight failed; keeping current VM running", "service", action.Service.Name, "error", err)
errs = append(errs, stageError(FailureStageVM, fmt.Errorf("preflight update %s: %w", action.Service.Name, err)))
continue
}
+ if r.settleRejections(&action, rejections) {
+ continue
+ }
prev := action.Service
if action.PreviousService != nil {
prev = *action.PreviousService
@@ -623,11 +628,15 @@ func (r *Reconciler) applyRolling(ctx context.Context, actions []Action) error {
for i, action := range updates {
r.logger.Info("updating service (stop + start)", "service", action.Service.Name)
- if err := r.preflight(ctx, action.Service); err != nil {
+ rejections, err := r.preflight(ctx, action.Service)
+ if err != nil {
r.logger.Error("volume preflight failed; keeping current VM running", "service", action.Service.Name, "error", err)
errs = append(errs, stageError(FailureStageVM, fmt.Errorf("preflight update %s: %w", action.Service.Name, err)))
break
}
+ if r.settleRejections(&action, rejections) {
+ continue
+ }
prev := action.Service
if action.PreviousService != nil {
prev = *action.PreviousService
@@ -657,11 +666,71 @@ func (r *Reconciler) applyRolling(ctx context.Context, actions []Action) error {
return nil
}
-func (r *Reconciler) preflight(ctx context.Context, svc config.ServiceConfig) error {
+func (r *Reconciler) preflight(ctx context.Context, svc config.ServiceConfig) ([]volume.Rejection, error) {
if manager, ok := r.vmManager.(volumePreflighter); ok {
return manager.Preflight(ctx, svc)
}
- return nil
+ return nil, nil
+}
+
+// settleRejections applies a preflight refusal to a planned update and reports
+// whether the update is now a no-op.
+//
+// This is what makes an advisory preflight rejection *terminal*. The VM is
+// still live at this point, and once the clamp lands the desired configuration
+// no longer differs from the running one, so no further update is planned —
+// not merely no further failure. Without it the measurement repeats on every
+// tick forever.
+func (r *Reconciler) settleRejections(action *Action, rejections []volume.Rejection) bool {
+ if !clampRejected(&action.Service, rejections) {
+ return false
+ }
+ instance := r.vmManager.List()[action.Service.Name]
+ if instance == nil || needsUpdate(instance, action.Service) {
+ // Something else about the service still differs — an image change,
+ // for instance — so the update proceeds, now carrying the effective
+ // volume size and performing no resize.
+ return false
+ }
+ r.logger.Info("volume size request refused; keeping the effective size and skipping the update",
+ "service", action.Service.Name)
+ return true
+}
+
+// clampRejected substitutes the effective size for every volume the preflight
+// refused, and reports whether anything changed.
+//
+// The refusal is applied by clamping rather than by failing the update, so an
+// update that changes the image *and* requests a refused shrink still deploys
+// the image. Failing the preflight would wedge every unrelated change behind a
+// size the node will never accept.
+func clampRejected(svc *config.ServiceConfig, rejections []volume.Rejection) bool {
+ if len(rejections) == 0 {
+ return false
+ }
+ byID := make(map[string]volume.Rejection, len(rejections))
+ for _, rejection := range rejections {
+ byID[rejection.LogicalID] = rejection
+ }
+ clamped := false
+ for i := range svc.Volumes {
+ rejection, ok := byID[svc.Name+"/"+svc.Volumes[i].Name]
+ if !ok {
+ continue
+ }
+ if svc.Volumes[i].SizeBytes == rejection.AppliedSizeBytes &&
+ svc.Volumes[i].ResizeGeneration == rejection.AppliedGeneration {
+ continue
+ }
+ // Substitute the whole effective configuration. Clamping only the size
+ // leaves the generation differing forever, and needsUpdate compares
+ // whole volume configs — so the update would be re-planned on every
+ // tick, which is worse than the single failure this replaces.
+ svc.Volumes[i].SizeBytes = rejection.AppliedSizeBytes
+ svc.Volumes[i].ResizeGeneration = rejection.AppliedGeneration
+ clamped = true
+ }
+ return clamped
}
// Reconcile is a convenience method that plans and applies in one step.
diff --git a/internal/reconciler/shrink_rejection_test.go b/internal/reconciler/shrink_rejection_test.go
new file mode 100644
index 0000000..aaeeb8d
--- /dev/null
+++ b/internal/reconciler/shrink_rejection_test.go
@@ -0,0 +1,124 @@
+package reconciler
+
+import (
+ "context"
+ "io"
+ "log/slog"
+ "testing"
+
+ "github.com/artemnikitin/firework/internal/config"
+ "github.com/artemnikitin/firework/internal/vm"
+ "github.com/artemnikitin/firework/internal/volume"
+)
+
+// rejectingVMManager refuses one volume's shrink at preflight, the way the
+// advisory pre-stop measurement does.
+type rejectingVMManager struct {
+ *fakeVMManager
+ rejections []volume.Rejection
+ preflightCalls int
+}
+
+func (f *rejectingVMManager) Preflight(context.Context, config.ServiceConfig) ([]volume.Rejection, error) {
+ f.preflightCalls++
+ return f.rejections, nil
+}
+
+func volumeService(name string, size int64, generation int64) config.ServiceConfig {
+ return config.ServiceConfig{
+ Name: name, Image: "/image", Kernel: "/kernel", VCPUs: 1, MemoryMB: 128,
+ Volumes: []config.VolumeConfig{{
+ Name: "data", Type: config.VolumeTypeLocal, MountPath: "/data",
+ SizeBytes: size, BoundNode: "node-1", ResizeGeneration: generation,
+ }},
+ }
+}
+
+func rejectingReconciler(t *testing.T, running config.ServiceConfig, applied int64) (*Reconciler, *rejectingVMManager) {
+ t.Helper()
+ manager := &rejectingVMManager{
+ fakeVMManager: newFakeVMManager(),
+ rejections: []volume.Rejection{{
+ LogicalID: running.Name + "/data", ResizeGeneration: 2, AppliedGeneration: 1,
+ RequestedSizeBytes: 2 * config.MiB, AppliedSizeBytes: applied, MinimumSizeBytes: 4 * config.MiB,
+ }},
+ }
+ manager.instances[running.Name] = &vm.Instance{Name: running.Name, State: vm.StateRunning, Config: running}
+ r := New(manager, slog.New(slog.NewTextHandler(io.Discard, nil)), nil, nil, "", 0).WithStateDir(t.TempDir())
+ return r, manager
+}
+
+// A preflight refusal is advisory and the VM is still live, so it must be made
+// terminal by clamping rather than by failing: the desired configuration stops
+// differing from the running one, so no further update is planned. Failing it
+// instead would re-measure on every tick forever.
+func TestPreflightRejectionIsTerminalAndLeavesTheVMRunning(t *testing.T) {
+ for _, strategy := range []string{"", "rolling"} {
+ name := strategy
+ if name == "" {
+ name = "all-at-once"
+ }
+ t.Run(name, func(t *testing.T) {
+ running := volumeService("app", 16*config.MiB, 1)
+ r, manager := rejectingReconciler(t, running, 16*config.MiB)
+ r.updateStrategy = strategy
+
+ // The desired revision asks for a shrink the node will refuse.
+ desired := config.NodeConfig{Node: "node-1", Services: []config.ServiceConfig{
+ volumeService("app", 2*config.MiB, 2),
+ }}
+
+ if err := r.Reconcile(context.Background(), desired); err != nil {
+ t.Fatalf("a refused shrink must not fail reconciliation: %v", err)
+ }
+ if len(manager.removeCalls) != 0 {
+ t.Fatalf("the VM was stopped for a refused shrink: %v", manager.removeCalls)
+ }
+ if len(manager.startCalls) != 0 {
+ t.Fatalf("the VM was restarted for a refused shrink: %v", manager.startCalls)
+ }
+ if manager.instances["app"].State != vm.StateRunning {
+ t.Fatal("the VM must stay running through a refused shrink")
+ }
+
+ // The following ticks must plan no further update. The agent-side
+ // normalization is what makes this hold in production; here the
+ // clamp inside the apply path already settles it.
+ for tick := 0; tick < 2; tick++ {
+ if err := r.Reconcile(context.Background(), desired); err != nil {
+ t.Fatalf("tick %d: %v", tick, err)
+ }
+ if len(manager.removeCalls) != 0 || len(manager.startCalls) != 0 {
+ t.Fatalf("tick %d stopped or restarted the service: removes=%v starts=%v",
+ tick, manager.removeCalls, manager.startCalls)
+ }
+ }
+ })
+ }
+}
+
+// An update that changes the image *and* requests a refused shrink must still
+// deploy the image. This falls out of clamping rather than blocking: failing
+// the preflight would wedge every unrelated change behind a refused size.
+func TestMixedUpdateStillDeploysTheImageWithNoResize(t *testing.T) {
+ running := volumeService("app", 16*config.MiB, 1)
+ r, manager := rejectingReconciler(t, running, 16*config.MiB)
+
+ updated := volumeService("app", 2*config.MiB, 2)
+ updated.Image = "/image-v2"
+ desired := config.NodeConfig{Node: "node-1", Services: []config.ServiceConfig{updated}}
+
+ if err := r.Reconcile(context.Background(), desired); err != nil {
+ t.Fatal(err)
+ }
+ if len(manager.startCalls) != 1 {
+ t.Fatalf("expected the image change to be deployed, got starts %v", manager.startCalls)
+ }
+ started := manager.instances["app"].Config
+ if started.Image != "/image-v2" {
+ t.Fatalf("the new image was not deployed: %q", started.Image)
+ }
+ if started.Volumes[0].SizeBytes != 16*config.MiB {
+ t.Fatalf("the refused size reached the launch path: %d", started.Volumes[0].SizeBytes)
+ }
+}
diff --git a/internal/scheduler/scheduler.go b/internal/scheduler/scheduler.go
index 9039370..f1a99f3 100644
--- a/internal/scheduler/scheduler.go
+++ b/internal/scheduler/scheduler.go
@@ -37,6 +37,95 @@ type StorageReservations struct {
SharedByBackend map[string]int64
RecordedLogicalIDs map[string]bool
SharedEnabled bool
+ // LocalUnknownByNode and SharedUnknownByBackend mark a scope whose
+ // remaining capacity cannot be proved because a retained record was only
+ // partially readable. Its lower bound is still charged above; the flag is
+ // what stops the unaccounted remainder from being handed out again.
+ LocalUnknownByNode map[string]bool
+ SharedUnknownByBackend map[string]bool
+ // LocalClassUnknown and SharedClassUnknown widen that block to a whole
+ // storage class, for a record so unreadable that no binding — and
+ // therefore no narrower scope — could be determined.
+ LocalClassUnknown bool
+ SharedClassUnknown bool
+ // UnknownCapacityKeys names one offending record per blocked scope, keyed
+ // by node ID, backend ID, or the storage class for a class-wide block.
+ // A block with no repair target named is one an operator cannot act on:
+ // the record is somewhere in the pool, and the message is all they have.
+ UnknownCapacityKeys map[string]string
+}
+
+// UnknownCapacityTarget returns a record key an operator should repair to lift
+// a scope's block, preferring the narrowest scope that names one.
+func (r StorageReservations) UnknownCapacityTarget(node, backend string, class config.VolumeType) string {
+ for _, scope := range []string{node, backend, string(class)} {
+ if scope == "" {
+ continue
+ }
+ if key := r.UnknownCapacityKeys[scope]; key != "" {
+ return key
+ }
+ }
+ return ""
+}
+
+// Pending reason codes. They are a bounded vocabulary because the status API,
+// fireworkctl, and the web UI all render them.
+const (
+ // ReasonInsufficientCompute means vCPU or memory, and nothing else.
+ ReasonInsufficientCompute = "insufficient_compute_capacity"
+ // ReasonVolumeCapacityUnavailable means the volume cannot bind to any
+ // candidate at all: no pool is configured there, or its retained binding
+ // names somewhere else. A configuration or placement fact.
+ ReasonVolumeCapacityUnavailable = "volume_capacity_unavailable"
+ // ReasonNodeStorageExhausted means the volume could bind, but the pool has
+ // no room for the new reservation. A capacity fact, resolved by freeing
+ // retained volumes or growing the pool.
+ ReasonNodeStorageExhausted = "node_storage_exhausted"
+ // ReasonStorageCapacityUnknown means remaining capacity cannot be proved,
+ // so new volume-bearing placement is withheld rather than guessed.
+ ReasonStorageCapacityUnknown = "storage_capacity_unknown"
+ // ReasonVolumeRecordInvalid means the service's own retained record could
+ // not be parsed, so it is not placed for the first time.
+ ReasonVolumeRecordInvalid = "volume_record_invalid"
+ // ReasonHostPortConflict means every candidate node already holds one of
+ // the service's (tcp, host_port) claims. It outranks the storage reasons
+ // below because the port check runs first: a node rejected on ports is
+ // never evaluated for storage, so a storage reason recorded elsewhere
+ // describes a different node than the one the operator has to fix.
+ ReasonHostPortConflict = "host_port_conflict"
+)
+
+// storageRank orders storage rejection causes from least to most actionable so
+// the dominant one survives across candidate nodes.
+func storageRank(reason string) int {
+ switch reason {
+ case ReasonVolumeCapacityUnavailable:
+ return 1
+ case ReasonStorageCapacityUnknown:
+ return 2
+ case ReasonNodeStorageExhausted:
+ return 3
+ default:
+ return 0
+ }
+}
+
+func storageReasonMessage(rejected storageRejection) string {
+ switch rejected.Reason {
+ case ReasonNodeStorageExhausted:
+ return "no active node has room for the requested volume reservation"
+ case ReasonStorageCapacityUnknown:
+ // The offending key is the whole value of this message. A cluster-wide
+ // block whose cause is only in the controller log leaves an operator
+ // grepping for an object they cannot name.
+ if rejected.Target != "" {
+ return fmt.Sprintf("remaining volume capacity cannot be verified; repair volume record %s", rejected.Target)
+ }
+ return "remaining volume capacity cannot be verified; repair the quarantined volume record"
+ default:
+ return "no active node satisfies volume binding and capacity"
+ }
}
type Pending struct {
@@ -205,7 +294,16 @@ func BuildNodeConfigs(assignment map[string][]config.ServiceConfig) []config.Nod
// guest destinations, so traffic silently reaches only one of them. A service
// is therefore placed only on a node where all of its claims are free, and its
// claims are taken atomically.
-func ScheduleWithStorage(services []config.ServiceConfig, nodes []Node, existing map[string]string, reservations StorageReservations) (map[string][]config.ServiceConfig, []Pending) {
+// pinned describes services the caller will render itself, outside this
+// scheduler, on a node it has already chosen. They still occupy node-exclusive
+// resources, so the scheduler has to be told about them or it will hand the
+// same host port to something else — the exact collision node-exclusive claims
+// exist to prevent. Compute is reserved by the caller adjusting node capacity;
+// ports cannot be expressed that way, so they are passed here.
+//
+// ScheduleWithStorage takes pinnedClaims as node -> claim -> holding service.
+// A nil map means nothing is pinned, which is the ordinary case.
+func ScheduleWithStorage(services []config.ServiceConfig, nodes []Node, existing map[string]string, reservations StorageReservations, pinnedClaims map[string]map[config.PortClaim]string) (map[string][]config.ServiceConfig, []Pending) {
result := make(map[string][]config.ServiceConfig, len(nodes))
usedVCPU := make(map[string]int, len(nodes))
usedMem := make(map[string]int, len(nodes))
@@ -218,6 +316,9 @@ func ScheduleWithStorage(services []config.ServiceConfig, nodes []Node, existing
result[node.InstanceID] = nil
groups[node.InstanceID] = make(map[string]bool)
claimedPorts[node.InstanceID] = make(map[config.PortClaim]string)
+ for claim, holder := range pinnedClaims[node.InstanceID] {
+ claimedPorts[node.InstanceID][claim] = holder
+ }
nodeByID[node.InstanceID] = node
}
@@ -281,6 +382,7 @@ func ScheduleWithStorage(services []config.ServiceConfig, nodes []Node, existing
claims := service.PortClaims()
chosen := ""
chosenService := service
+ var dominantStorage storageRejection
portConflict := ""
for _, node := range candidates {
if boundNode != "" && node.InstanceID != boundNode {
@@ -297,8 +399,14 @@ func ScheduleWithStorage(services []config.ServiceConfig, nodes []Node, existing
}
continue
}
- candidateService, localDelta, sharedDelta, ok := fitStorage(service, node, reservations, usedLocal, usedShared)
- if !ok {
+ candidateService, localDelta, sharedDelta, rejected := fitStorage(service, node, reservations, usedLocal, usedShared)
+ if rejected.Reason != "" {
+ // Keep the most actionable cause seen across candidates. The
+ // dominant reason tells the operator whether the placement is
+ // wrong or the chosen node is simply full.
+ if storageRank(rejected.Reason) > storageRank(dominantStorage.Reason) {
+ dominantStorage = rejected
+ }
continue
}
chosen = node.InstanceID
@@ -310,14 +418,14 @@ func ScheduleWithStorage(services []config.ServiceConfig, nodes []Node, existing
break
}
if chosen == "" {
- reason := "insufficient_compute_capacity"
+ reason := ReasonInsufficientCompute
message := "no active node satisfies compute capacity"
- if len(service.Volumes) > 0 {
- reason = "volume_capacity_unavailable"
- message = "no active node satisfies volume binding and capacity"
+ if dominantStorage.Reason != "" {
+ reason = dominantStorage.Reason
+ message = storageReasonMessage(dominantStorage)
}
if portConflict != "" {
- reason = "host_port_conflict"
+ reason = ReasonHostPortConflict
message = portConflict
}
pending = append(pending, Pending{Service: service.Name, ReasonCode: reason, Message: message})
@@ -374,7 +482,19 @@ func hasSharedVolume(service config.ServiceConfig) bool {
return false
}
-func fitStorage(service config.ServiceConfig, node Node, reservations StorageReservations, usedLocal, usedShared map[string]int64) (config.ServiceConfig, int64, int64, bool) {
+// fitStorage reports whether a service's volumes can bind to a node, and why
+// not when they cannot. The reason separates a placement fact (the volume
+// cannot bind here at all) from a capacity fact (it could bind, but the pool
+// has no room), because the two have opposite operator remedies.
+// storageRejection is why a node was refused, and where to look to fix it.
+type storageRejection struct {
+ Reason string
+ // Target names a quarantined record to repair, for the reasons where one
+ // exists. Empty otherwise.
+ Target string
+}
+
+func fitStorage(service config.ServiceConfig, node Node, reservations StorageReservations, usedLocal, usedShared map[string]int64) (config.ServiceConfig, int64, int64, storageRejection) {
candidate := service
candidate.Volumes = append([]config.VolumeConfig(nil), service.Volumes...)
var localDelta, sharedDelta int64
@@ -384,7 +504,7 @@ func fitStorage(service config.ServiceConfig, node Node, reservations StorageRes
switch volume.Type {
case config.VolumeTypeLocal:
if node.LocalCapacityBytes <= 0 || (volume.BoundNode != "" && volume.BoundNode != node.InstanceID) {
- return service, 0, 0, false
+ return service, 0, 0, storageRejection{Reason: ReasonVolumeCapacityUnavailable}
}
volume.BoundNode = node.InstanceID
if !reservations.RecordedLogicalIDs[logicalID] {
@@ -392,7 +512,7 @@ func fitStorage(service config.ServiceConfig, node Node, reservations StorageRes
}
case config.VolumeTypeShared:
if node.SharedBackendID == "" || (volume.SharedBackendID != "" && volume.SharedBackendID != node.SharedBackendID) {
- return service, 0, 0, false
+ return service, 0, 0, storageRejection{Reason: ReasonVolumeCapacityUnavailable}
}
volume.SharedBackendID = node.SharedBackendID
if !reservations.RecordedLogicalIDs[logicalID] {
@@ -400,11 +520,33 @@ func fitStorage(service config.ServiceConfig, node Node, reservations StorageRes
}
}
}
- if reservations.LocalByNode[node.InstanceID]+usedLocal[node.InstanceID]+localDelta > node.LocalCapacityBytes {
- return service, 0, 0, false
+ // A service that adds no new local reservation cannot recover capacity by
+ // being rejected, it can only be evicted. Volumes already counted in
+ // LocalByNode contribute a zero delta, and a service with no volumes at
+ // all contributes nothing — so retained reservations above the pool must
+ // not make the node reject either of them. Only a genuinely new
+ // allocation is checked against the pool.
+ if localDelta > 0 && reservations.LocalByNode[node.InstanceID]+usedLocal[node.InstanceID]+localDelta > node.LocalCapacityBytes {
+ return service, 0, 0, storageRejection{Reason: ReasonNodeStorageExhausted}
}
if sharedDelta > 0 && node.SharedCapacityBytes > 0 && reservations.SharedByBackend[node.SharedBackendID]+usedShared[node.SharedBackendID]+sharedDelta > node.SharedCapacityBytes {
- return service, 0, 0, false
+ return service, 0, 0, storageRejection{Reason: ReasonNodeStorageExhausted}
+ }
+ // A quarantined record whose reservation could not be read makes the
+ // node's remaining pool unknowable. New volume-bearing placement is
+ // withheld there rather than allocated against capacity that may already
+ // be occupied; an already-placed service is re-rendered untouched.
+ if localDelta > 0 && (reservations.LocalUnknownByNode[node.InstanceID] || reservations.LocalClassUnknown) {
+ return service, 0, 0, storageRejection{
+ Reason: ReasonStorageCapacityUnknown,
+ Target: reservations.UnknownCapacityTarget(node.InstanceID, "", config.VolumeTypeLocal),
+ }
+ }
+ if sharedDelta > 0 && (reservations.SharedUnknownByBackend[node.SharedBackendID] || reservations.SharedClassUnknown) {
+ return service, 0, 0, storageRejection{
+ Reason: ReasonStorageCapacityUnknown,
+ Target: reservations.UnknownCapacityTarget("", node.SharedBackendID, config.VolumeTypeShared),
+ }
}
- return candidate, localDelta, sharedDelta, true
+ return candidate, localDelta, sharedDelta, storageRejection{}
}
diff --git a/internal/scheduler/scheduler_test.go b/internal/scheduler/scheduler_test.go
index ec41d9f..3fc2f6b 100644
--- a/internal/scheduler/scheduler_test.go
+++ b/internal/scheduler/scheduler_test.go
@@ -255,7 +255,7 @@ func TestScheduleWithStorageBindsLocalVolumeAndHonorsRetainedBinding(t *testing.
{InstanceID: "small", CapacityVCPUs: 4, CapacityMemMB: 1024, LocalCapacityBytes: 5 * config.GiB},
{InstanceID: "large", CapacityVCPUs: 4, CapacityMemMB: 1024, LocalCapacityBytes: 20 * config.GiB},
}
- result, pending := ScheduleWithStorage([]config.ServiceConfig{service}, nodes, nil, StorageReservations{})
+ result, pending := ScheduleWithStorage([]config.ServiceConfig{service}, nodes, nil, StorageReservations{}, nil)
if len(pending) != 0 || len(result["large"]) != 1 {
t.Fatalf("unexpected placement result=%#v pending=%#v", result, pending)
}
@@ -264,7 +264,7 @@ func TestScheduleWithStorageBindsLocalVolumeAndHonorsRetainedBinding(t *testing.
}
service.Volumes[0].BoundNode = "lost"
- _, pending = ScheduleWithStorage([]config.ServiceConfig{service}, nodes, nil, StorageReservations{})
+ _, pending = ScheduleWithStorage([]config.ServiceConfig{service}, nodes, nil, StorageReservations{}, nil)
if len(pending) != 1 || pending[0].ReasonCode != "local_volume_node_unavailable" {
t.Fatalf("unexpected retained binding result: %#v", pending)
}
@@ -274,7 +274,7 @@ func TestScheduleWithStorageKeepsSharedPendingUntilSafetyGate(t *testing.T) {
service := svc("db", 1, 256)
service.Volumes = []config.VolumeConfig{{Name: "data", Type: config.VolumeTypeShared, MountPath: "/data", SizeBytes: config.GiB}}
nodes := []Node{{InstanceID: "node", CapacityVCPUs: 4, CapacityMemMB: 1024, SharedBackendID: "primary"}}
- _, pending := ScheduleWithStorage([]config.ServiceConfig{service}, nodes, nil, StorageReservations{})
+ _, pending := ScheduleWithStorage([]config.ServiceConfig{service}, nodes, nil, StorageReservations{}, nil)
if len(pending) != 1 || pending[0].ReasonCode != "shared_volume_runtime_unavailable" {
t.Fatalf("unexpected pending result: %#v", pending)
}
@@ -309,7 +309,7 @@ func TestScheduleWithStorageSeparatesServicesSharingHostPort(t *testing.T) {
}
nodes := []Node{node("i-001", 32, 16384), node("i-002", 4, 2048)}
- result, pending := ScheduleWithStorage(services, nodes, nil, StorageReservations{})
+ result, pending := ScheduleWithStorage(services, nodes, nil, StorageReservations{}, nil)
if len(pending) != 0 {
t.Fatalf("unexpected pending services: %#v", pending)
}
@@ -327,7 +327,7 @@ func TestScheduleWithStorageKeepsRepeatedHostPortsOnDifferentNodes(t *testing.T)
nodes := []Node{node("i-001", 8, 4096), node("i-002", 8, 4096)}
existing := map[string]string{"a": "i-001", "b": "i-002"}
- result, pending := ScheduleWithStorage(services, nodes, existing, StorageReservations{})
+ result, pending := ScheduleWithStorage(services, nodes, existing, StorageReservations{}, nil)
if len(pending) != 0 {
t.Fatalf("unexpected pending services: %#v", pending)
}
@@ -343,7 +343,7 @@ func TestScheduleWithStorageLeavesConflictingServicePending(t *testing.T) {
}
nodes := []Node{node("i-001", 8, 4096)}
- result, pending := ScheduleWithStorage(services, nodes, nil, StorageReservations{})
+ result, pending := ScheduleWithStorage(services, nodes, nil, StorageReservations{}, nil)
if len(result["i-001"]) != 1 {
t.Fatalf("expected exactly one service placed, got %#v", result)
}
@@ -367,7 +367,7 @@ func TestScheduleWithStorageRelocatesExistingPlacementOnNewConflict(t *testing.T
nodes := []Node{node("i-001", 8, 4096), node("i-002", 8, 4096)}
existing := map[string]string{"a": "i-001", "b": "i-001"}
- result, pending := ScheduleWithStorage(services, nodes, existing, StorageReservations{})
+ result, pending := ScheduleWithStorage(services, nodes, existing, StorageReservations{}, nil)
if len(pending) != 0 {
t.Fatalf("unexpected pending services: %#v", pending)
}
@@ -389,7 +389,7 @@ func TestScheduleWithStorageTreatsMultipleClaimsAtomically(t *testing.T) {
nodes := []Node{node("i-001", 16, 8192), node("i-002", 8, 4096)}
existing := map[string]string{"keeper": "i-001"}
- result, pending := ScheduleWithStorage(services, nodes, existing, StorageReservations{})
+ result, pending := ScheduleWithStorage(services, nodes, existing, StorageReservations{}, nil)
if len(pending) != 0 {
t.Fatalf("unexpected pending services: %#v", pending)
}
@@ -408,7 +408,7 @@ func TestScheduleWithStorageRejectsSelfConflictingService(t *testing.T) {
service := withPorts("broken", 2, 512, 8080, 8080)
nodes := []Node{node("i-001", 8, 4096), node("i-002", 8, 4096)}
- result, pending := ScheduleWithStorage([]config.ServiceConfig{service}, nodes, nil, StorageReservations{})
+ result, pending := ScheduleWithStorage([]config.ServiceConfig{service}, nodes, nil, StorageReservations{}, nil)
if len(pending) != 1 || pending[0].ReasonCode != "duplicate_host_port_claims" {
t.Fatalf("unexpected pending result: %#v", pending)
}
diff --git a/internal/scheduler/volume_hardening_test.go b/internal/scheduler/volume_hardening_test.go
new file mode 100644
index 0000000..6509cb5
--- /dev/null
+++ b/internal/scheduler/volume_hardening_test.go
@@ -0,0 +1,174 @@
+package scheduler
+
+import (
+ "testing"
+
+ "github.com/artemnikitin/firework/internal/config"
+)
+
+// exhaustedNode has a pool whose retained reservations already exceed its
+// configured capacity — the state an oversized `size:` edit produces, and the
+// one that used to make the node reject every workload on it.
+func exhaustedNode() ([]Node, StorageReservations) {
+ nodes := []Node{{
+ InstanceID: "i-1", CapacityVCPUs: 8, CapacityMemMB: 8192,
+ LocalCapacityBytes: 100 * config.MiB,
+ }}
+ reservations := StorageReservations{
+ LocalByNode: map[string]int64{"i-1": 500 * config.MiB},
+ SharedByBackend: map[string]int64{},
+ RecordedLogicalIDs: map[string]bool{"kept/data": true},
+ }
+ return nodes, reservations
+}
+
+func localVolumeService(name string, size int64) config.ServiceConfig {
+ return config.ServiceConfig{
+ Name: name, VCPUs: 1, MemoryMB: 512,
+ Volumes: []config.VolumeConfig{{
+ Name: "data", Type: config.VolumeTypeLocal, MountPath: "/data", SizeBytes: size,
+ }},
+ }
+}
+
+func pendingReason(pending []Pending, service string) string {
+ for _, item := range pending {
+ if item.Service == service {
+ return item.ReasonCode
+ }
+ }
+ return ""
+}
+
+func placedOn(assignments map[string][]config.ServiceConfig, service string) string {
+ for node, services := range assignments {
+ for _, placed := range services {
+ if placed.Name == service {
+ return node
+ }
+ }
+ }
+ return ""
+}
+
+// A node whose retained reservations exceed its pool must still run the
+// workloads that do not add to those reservations. Rejecting them cannot
+// recover a single byte; it can only evict.
+func TestOverReservedNodeStillAcceptsServicesThatAddNoReservation(t *testing.T) {
+ nodes, reservations := exhaustedNode()
+ services := []config.ServiceConfig{
+ {Name: "stateless", VCPUs: 1, MemoryMB: 512},
+ localVolumeService("kept", 16*config.MiB),
+ }
+ existing := map[string]string{"stateless": "i-1", "kept": "i-1"}
+
+ assignments, pending := ScheduleWithStorage(services, nodes, existing, reservations, nil)
+
+ if len(pending) != 0 {
+ t.Fatalf("expected no pending services, got %#v", pending)
+ }
+ if placedOn(assignments, "stateless") != "i-1" || placedOn(assignments, "kept") != "i-1" {
+ t.Fatalf("expected both services to stay on i-1, got %#v", assignments)
+ }
+}
+
+// The capacity guard still holds for a genuinely new allocation, and now says
+// which of the two very different storage causes applied.
+func TestStorageRejectionReasonsAreDistinct(t *testing.T) {
+ nodes, reservations := exhaustedNode()
+ // An active node with no local pool at all: a volume bound there cannot
+ // bind, which is a placement fact rather than a capacity one.
+ nodes = append(nodes, Node{InstanceID: "i-poolless", CapacityVCPUs: 8, CapacityMemMB: 8192})
+
+ tests := []struct {
+ name string
+ service config.ServiceConfig
+ want string
+ }{
+ {
+ name: "new allocation on a full pool",
+ service: localVolumeService("fresh", 16*config.MiB),
+ want: ReasonNodeStorageExhausted,
+ },
+ {
+ name: "bound node has no local pool configured",
+ service: func() config.ServiceConfig {
+ svc := localVolumeService("elsewhere", 16*config.MiB)
+ svc.Volumes[0].BoundNode = "i-poolless"
+ return svc
+ }(),
+ want: ReasonVolumeCapacityUnavailable,
+ },
+ {
+ name: "compute only",
+ service: config.ServiceConfig{Name: "huge", VCPUs: 64, MemoryMB: 65536},
+ want: ReasonInsufficientCompute,
+ },
+ }
+
+ for _, test := range tests {
+ t.Run(test.name, func(t *testing.T) {
+ _, pending := ScheduleWithStorage([]config.ServiceConfig{test.service}, nodes, nil, reservations, nil)
+ if got := pendingReason(pending, test.service.Name); got != test.want {
+ t.Fatalf("expected reason %q, got %q (%#v)", test.want, got, pending)
+ }
+ })
+ }
+}
+
+// A quarantined record whose reservation could not be read makes the node's
+// remaining pool unknowable, so new volume-bearing placement waits for the
+// repair rather than being allocated against capacity that may be occupied.
+// Existing placements and stateless services are untouched.
+func TestUnknownCapacityBlocksOnlyNewVolumeBearingPlacement(t *testing.T) {
+ nodes, reservations := exhaustedNode()
+ reservations.LocalByNode = map[string]int64{"i-1": 16 * config.MiB}
+ reservations.LocalUnknownByNode = map[string]bool{"i-1": true}
+
+ services := []config.ServiceConfig{
+ {Name: "stateless", VCPUs: 1, MemoryMB: 512},
+ localVolumeService("kept", 16*config.MiB),
+ localVolumeService("fresh", 16*config.MiB),
+ }
+ assignments, pending := ScheduleWithStorage(services, nodes, map[string]string{"kept": "i-1"}, reservations, nil)
+
+ if placedOn(assignments, "stateless") == "" || placedOn(assignments, "kept") == "" {
+ t.Fatalf("existing and stateless workloads must keep running: %#v (%#v)", assignments, pending)
+ }
+ if got := pendingReason(pending, "fresh"); got != ReasonStorageCapacityUnknown {
+ t.Fatalf("expected %q for a new volume-bearing service, got %q", ReasonStorageCapacityUnknown, got)
+ }
+}
+
+// The eviction path end to end: an oversized size: edit raises the node's
+// reservations above its pool, and every service on it must still be rendered.
+// BuildNodeConfigs dropping a node's services is what the agent turns into a
+// delete for each one, so an empty render here is the eviction itself.
+func TestOversizedReservationDoesNotEmptyTheRenderedNodeConfig(t *testing.T) {
+ nodes := []Node{{
+ InstanceID: "i-1", CapacityVCPUs: 8, CapacityMemMB: 8192,
+ LocalCapacityBytes: 100 * config.MiB,
+ }}
+ // The operator edited size: to something far above the pool; the record
+ // was already retained, so it contributes its inflated reservation.
+ reservations := StorageReservations{
+ LocalByNode: map[string]int64{"i-1": 900 * config.MiB},
+ SharedByBackend: map[string]int64{},
+ RecordedLogicalIDs: map[string]bool{"db/data": true},
+ }
+ services := []config.ServiceConfig{
+ {Name: "web", VCPUs: 1, MemoryMB: 512},
+ {Name: "api", VCPUs: 1, MemoryMB: 512},
+ localVolumeService("db", 900*config.MiB),
+ }
+ existing := map[string]string{"web": "i-1", "api": "i-1", "db": "i-1"}
+
+ assignments, pending := ScheduleWithStorage(services, nodes, existing, reservations, nil)
+ if len(pending) != 0 {
+ t.Fatalf("expected no service to be evicted, got %#v", pending)
+ }
+ rendered := BuildNodeConfigs(assignments)
+ if len(rendered) != 1 || len(rendered[0].Services) != 3 {
+ t.Fatalf("expected all three services rendered for i-1, got %#v", rendered)
+ }
+}
diff --git a/internal/statusmodel/status.go b/internal/statusmodel/status.go
index 234b704..4d6be0b 100644
--- a/internal/statusmodel/status.go
+++ b/internal/statusmodel/status.go
@@ -63,7 +63,13 @@ var (
"ConfigFetched", "ConfigParsed", "NetworkReady", "CapacityReady",
"ImagesReady", "VMsReconciled", "Reconciled", "LocalRoutesReady",
}
- nonBlockingConditionTypes = []string{"PeerRoutesReady"}
+ // VolumeSizesApplied is false while this node is running a volume at a
+ // size other than the one the desired revision asked for. It is
+ // non-blocking because the workload is healthy — it is running, just not
+ // at the requested quota — but it must not read as ordinary convergence,
+ // or the operator sees a service quietly running at the wrong size with no
+ // explanation.
+ nonBlockingConditionTypes = []string{"PeerRoutesReady", "VolumeSizesApplied"}
)
// BlockingConditionTypes returns the conditions whose failure is fatal.
@@ -152,6 +158,13 @@ type VolumeStatus struct {
ResizeGeneration int64 `json:"resize_generation,omitempty"`
State string `json:"state"`
LastError string `json:"last_error,omitempty"`
+ // RequestedSizeBytes is what the desired revision asked for, when that
+ // differs from the effective DesiredSizeBytes the cluster accepted and
+ // rendered. Equal sizes are reported only through DesiredSizeBytes, so an
+ // unrejected volume's surface is unchanged.
+ RequestedSizeBytes int64 `json:"requested_size_bytes,omitempty"`
+ Rejected bool `json:"rejected,omitempty"`
+ RejectedReason string `json:"rejected_reason,omitempty"`
}
type AgentStatus struct {
diff --git a/internal/vm/manager.go b/internal/vm/manager.go
index 5f15078..5ff4d68 100644
--- a/internal/vm/manager.go
+++ b/internal/vm/manager.go
@@ -51,8 +51,34 @@ const (
// StateRecoveryPending means durable state exists but ownership could not
// be proved. Firework preserves the process and files and blocks duplicates.
StateRecoveryPending State = "recovery_pending"
+ // StateStarting is published while a start has released the manager lock
+ // to prepare volumes. It exists so List reports something truthful during
+ // a multi-minute mkfs or resize rather than reporting nothing at all.
+ StateStarting State = "starting"
+ // StateStartAborting means a Stop or Remove arrived while a start was in
+ // its unlocked preparation phase. The start's own final phase observes it
+ // and cleans up without launching anything.
+ StateStartAborting State = "start_aborting"
)
+var (
+ // ErrStartAborted reports that a start was cancelled by a concurrent Stop
+ // or Remove before anything was launched. It is a benign race, not a
+ // fault: the caller must retry rather than record a reconcile failure.
+ ErrStartAborted = errors.New("start aborted by a concurrent stop or remove")
+ // ErrStartInProgress reports that another start for the same service is
+ // still in its preparation phase. Like ErrStartAborted this is a retry
+ // signal, not a failure.
+ ErrStartInProgress = errors.New("start already in progress")
+)
+
+// IsStartRace reports whether an error is one of the benign start-barrier
+// races. Callers use it to classify a reconciliation as incomplete — retry on
+// the next tick without advancing the applied revision — rather than failed.
+func IsStartRace(err error) bool {
+ return errors.Is(err, ErrStartAborted) || errors.Is(err, ErrStartInProgress)
+}
+
// Instance represents a running Firecracker microVM.
type Instance struct {
// Name is the service name from the config.
@@ -72,6 +98,10 @@ type Instance struct {
instanceID string
manifest *instanceManifest
+ // startID identifies one Start attempt. Phase 3 validates against it
+ // rather than against the service name, so a placeholder that was cleared
+ // and replaced by a later attempt is never mistaken for one's own.
+ startID string
}
// Manager manages the lifecycle of Firecracker microVMs on the local host.
@@ -119,23 +149,54 @@ func NewManagerWithVolumes(firecrackerBin, stateDir string, logger *slog.Logger,
// Preflight validates persistent volumes without changing them. Reconciliation
// calls this before stopping an existing VM so a failed resize leaves it live.
-func (m *Manager) Preflight(ctx context.Context, svc config.ServiceConfig) error {
+// It also returns any size requests it refused. A rejection never goes through
+// setVolumeError: it is a decision rather than a fault, and recording it there
+// would set a volume_failed reason code and trigger the blanket overwrite that
+// relabels every one of the service's volumes as errored.
+func (m *Manager) Preflight(ctx context.Context, svc config.ServiceConfig) ([]volume.Rejection, error) {
if len(svc.Volumes) == 0 {
m.setVolumeError(svc.Name, nil)
- return nil
+ return nil, nil
}
if err := validateVolumeKernelArgs(svc); err != nil {
m.setVolumeError(svc.Name, err)
- return err
+ return nil, err
}
if m.volumeManager == nil {
err := fmt.Errorf("service %s declares volumes but agent storage is not configured", svc.Name)
m.setVolumeError(svc.Name, err)
- return err
+ return nil, err
}
- err := m.volumeManager.Preflight(ctx, svc)
+ rejections, err := m.volumeManager.Preflight(ctx, svc)
m.setVolumeError(svc.Name, err)
- return err
+ return rejections, err
+}
+
+// VolumeRejections returns the agent's current per-volume refusal snapshot.
+func (m *Manager) VolumeRejections() map[string]volume.Rejection {
+ if m.volumeManager == nil {
+ return nil
+ }
+ return m.volumeManager.Rejections()
+}
+
+// SeedVolumeRejectionsForTest installs a refusal snapshot without running a
+// real filesystem operation, so the status and convergence paths can be
+// exercised without a live pool.
+func (m *Manager) SeedVolumeRejectionsForTest(rejections map[string]volume.Rejection) {
+ if m.volumeManager == nil {
+ return
+ }
+ m.volumeManager.SeedRejectionsForTest(rejections)
+}
+
+// NormalizeVolumes clamps a desired node configuration to the sizes the node
+// is actually able to serve, before anything else in the tick reads it.
+func (m *Manager) NormalizeVolumes(services []config.ServiceConfig) {
+ if m.volumeManager == nil {
+ return
+ }
+ m.volumeManager.NormalizeVolumes(services)
}
// VolumeError returns the latest persistent-volume preparation failure for a
@@ -146,6 +207,32 @@ func (m *Manager) VolumeError(service string) string {
return m.volumeErrors[service]
}
+// clampToPrepared substitutes the effective configuration — size and
+// generation — for every volume whose request was refused, so the instance the
+// next tick compares against describes what is actually running. The refused
+// request is reported from the volume manager's rejection snapshot instead,
+// which is what the control-plane acknowledgement matches on.
+func clampToPrepared(svc config.ServiceConfig, prepared []volume.PreparedVolume) config.ServiceConfig {
+ effective := make(map[string]volume.PreparedVolume, len(prepared))
+ for _, preparedVolume := range prepared {
+ if preparedVolume.Rejected {
+ effective[preparedVolume.LogicalID] = preparedVolume
+ }
+ }
+ if len(effective) == 0 {
+ return svc
+ }
+ clamped := svc
+ clamped.Volumes = append([]config.VolumeConfig(nil), svc.Volumes...)
+ for i := range clamped.Volumes {
+ if preparedVolume, ok := effective[svc.Name+"/"+clamped.Volumes[i].Name]; ok {
+ clamped.Volumes[i].SizeBytes = preparedVolume.SizeBytes
+ clamped.Volumes[i].ResizeGeneration = preparedVolume.ResizeGeneration
+ }
+ }
+ return clamped
+}
+
func (m *Manager) setVolumeError(service string, err error) {
m.mu.Lock()
defer m.mu.Unlock()
@@ -156,32 +243,64 @@ func (m *Manager) setVolumeError(service string, err error) {
m.volumeErrors[service] = err.Error()
}
-func validateVolumeKernelArgs(svc config.ServiceConfig) error {
- volumes := append([]config.VolumeConfig(nil), svc.Volumes...)
- sort.Slice(volumes, func(i, j int) bool { return volumes[i].Name < volumes[j].Name })
- guestVolumes := make([]guestVolume, 0, len(volumes))
- for i, volume := range volumes {
+// defaultKernelArgs is the boot command line used when a service declares none.
+const defaultKernelArgs = "console=ttyS0 reboot=k panic=1 pci=off"
+
+// buildBootArgs composes a service's kernel command line and enforces the
+// command-line length limit.
+//
+// It is the single place boot args are built. Preflight's early check and the
+// launch path previously constructed the payload separately and drifted: the
+// length check lived only in the update path, so ActionCreate could boot a VM
+// with an over-long command line instead of failing with a clear error.
+func buildBootArgs(svc config.ServiceConfig, guestVolumes []guestVolume) (string, error) {
+ kernelArgs := svc.KernelArgs
+ if kernelArgs == "" {
+ kernelArgs = defaultKernelArgs
+ }
+ if len(guestVolumes) > 0 {
+ payload, err := json.Marshal(guestVolumePayload{Version: 1, Volumes: guestVolumes})
+ if err != nil {
+ return "", fmt.Errorf("marshal guest volume payload: %w", err)
+ }
+ arg := "firework.volumes64=" + base64.RawURLEncoding.EncodeToString(payload)
+ kernelArgs = insertBeforeApplicationSeparator(kernelArgs, arg)
+ }
+ if len(kernelArgs) > maxKernelCommandLineBytes {
+ what := "kernel command line"
+ if len(guestVolumes) > 0 {
+ what = "kernel command line with volume payload"
+ }
+ return "", fmt.Errorf("service %s: %s is %d bytes; maximum is %d", svc.Name, what, len(kernelArgs), maxKernelCommandLineBytes)
+ }
+ return kernelArgs, nil
+}
+
+// guestVolumesFromConfig builds the guest payload entries from declared
+// volumes, for the preflight that runs before anything has been prepared.
+func guestVolumesFromConfig(volumes []config.VolumeConfig) ([]guestVolume, error) {
+ ordered := append([]config.VolumeConfig(nil), volumes...)
+ sort.Slice(ordered, func(i, j int) bool { return ordered[i].Name < ordered[j].Name })
+ guestVolumes := make([]guestVolume, 0, len(ordered))
+ for i, declared := range ordered {
device, err := guestBlockDevice(i)
if err != nil {
- return err
+ return nil, err
}
guestVolumes = append(guestVolumes, guestVolume{
- Name: volume.Name, Device: device, MountPath: volume.MountPath, Type: volume.Type,
+ Name: declared.Name, Device: device, MountPath: declared.MountPath, Type: declared.Type,
})
}
- payload, err := json.Marshal(guestVolumePayload{Version: 1, Volumes: guestVolumes})
+ return guestVolumes, nil
+}
+
+func validateVolumeKernelArgs(svc config.ServiceConfig) error {
+ guestVolumes, err := guestVolumesFromConfig(svc.Volumes)
if err != nil {
- return fmt.Errorf("marshal guest volume payload: %w", err)
- }
- kernelArgs := svc.KernelArgs
- if kernelArgs == "" {
- kernelArgs = "console=ttyS0 reboot=k panic=1 pci=off"
- }
- kernelArgs = insertBeforeApplicationSeparator(kernelArgs, "firework.volumes64="+base64.RawURLEncoding.EncodeToString(payload))
- if len(kernelArgs) > maxKernelCommandLineBytes {
- return fmt.Errorf("service %s: kernel command line with volume payload is %d bytes; maximum is %d", svc.Name, len(kernelArgs), maxKernelCommandLineBytes)
+ return err
}
- return nil
+ _, err = buildBootArgs(svc, guestVolumes)
+ return err
}
// List returns a snapshot of all known VM instances.
@@ -198,59 +317,155 @@ func (m *Manager) List() map[string]*Instance {
}
// Start launches a new Firecracker microVM for the given service config.
+// Start launches a microVM for a service.
+//
+// It runs in three phases so the manager lock is not held across volume
+// preparation, which can spend minutes in mkfs.ext4, e2fsck, or resize2fs.
+// Holding the lock there stalled every reader of it — including the heartbeat
+// goroutine, which reaches it through List — so a node went stale precisely
+// while it was busy resizing its own services' volumes.
+//
+// Releasing the lock opens a window in which a Stop or Remove can arrive, so
+// the phases are governed by a barrier rather than by extra branches:
+//
+// absent -> StateStarting phase 1 publishes the placeholder
+// StateStarting -> StateRunning phase 3, own startID still present
+// StateStarting -> StateStartAborting Stop or Remove during phase 2
+// StateStartAborting -> absent phase 3 observes the abort
+// StateStarting -> absent phase 2 failed
+//
+// Phase 3 confirms ownership before any side effect: no manifest is written
+// and nothing is launched unless the placeholder is still this attempt's own.
func (m *Manager) Start(ctx context.Context, svc config.ServiceConfig) error {
+ startID, vmDir, socketPath, err := m.beginStart(svc)
+ if err != nil {
+ return err
+ }
+
+ // Phase 2 runs without the manager lock. Its side effects — a created or
+ // resized volume image — are durable, retained state by design, so they
+ // are deliberately not rolled back when the start is aborted: the next
+ // start reuses them.
+ prepared, err := m.prepareVolumes(ctx, svc)
+ if err != nil {
+ m.discardStart(svc.Name, startID)
+ return err
+ }
+
+ return m.finishStart(ctx, svc, startID, vmDir, socketPath, prepared)
+}
+
+// beginStart is phase 1: it takes the entry checks and publishes the starting
+// placeholder under the manager lock.
+func (m *Manager) beginStart(svc config.ServiceConfig) (startID, vmDir, socketPath string, err error) {
m.mu.Lock()
defer m.mu.Unlock()
if inst, exists := m.instances[svc.Name]; exists {
- if inst.State == StateRecoveryPending {
- return fmt.Errorf("service %s has ambiguous surviving state: %s", svc.Name, inst.LastError)
- }
- if inst.State == StateRunning || inst.State == StateStopping {
- return fmt.Errorf("service %s is already active (pid %d, state %s)", svc.Name, inst.PID, inst.State)
+ switch inst.State {
+ case StateRecoveryPending:
+ return "", "", "", fmt.Errorf("service %s has ambiguous surviving state: %s", svc.Name, inst.LastError)
+ case StateRunning, StateStopping:
+ return "", "", "", fmt.Errorf("service %s is already active (pid %d, state %s)", svc.Name, inst.PID, inst.State)
+ case StateStarting, StateStartAborting:
+ // A start already holds this name. Rejecting here is what keeps
+ // the agent API and shutdown paths from racing the reconcile loop.
+ return "", "", "", fmt.Errorf("service %s is in state %s: %w", svc.Name, inst.State, ErrStartInProgress)
}
}
m.logger.Info("starting microVM", "service", svc.Name, "vcpus", svc.VCPUs, "memory_mb", svc.MemoryMB)
- vmDir := filepath.Join(m.stateDir, "vms", svc.Name)
+ vmDir = filepath.Join(m.stateDir, "vms", svc.Name)
if err := m.reclaimUnownedState(svc.Name, vmDir); err != nil {
- return err
+ return "", "", "", err
}
if err := os.MkdirAll(vmDir, 0o755); err != nil {
- return fmt.Errorf("creating vm dir: %w", err)
+ return "", "", "", fmt.Errorf("creating vm dir: %w", err)
}
- socketPath := filepath.Join(vmDir, "firecracker.sock")
+ socketPath = filepath.Join(vmDir, "firecracker.sock")
// Remove stale socket if it exists.
_ = os.Remove(socketPath)
- var prepared []volume.PreparedVolume
- var err error
- if len(svc.Volumes) > 0 {
- if m.volumeManager == nil {
- err := fmt.Errorf("service %s declares volumes but agent storage is not configured", svc.Name)
- m.volumeErrors[svc.Name] = err.Error()
- return err
- }
- prepared, err = m.volumeManager.Prepare(ctx, svc)
- if err != nil {
- m.volumeErrors[svc.Name] = err.Error()
- return fmt.Errorf("preparing volumes: %w", err)
+ startID, err = newInstanceID()
+ if err != nil {
+ return "", "", "", err
+ }
+ m.instances[svc.Name] = &Instance{
+ Name: svc.Name, Config: svc, State: StateStarting,
+ SocketPath: socketPath, startID: startID,
+ }
+ return startID, vmDir, socketPath, nil
+}
+
+// prepareVolumes is phase 2. It runs without the manager lock, so it records
+// volume errors through setVolumeError rather than writing the map directly.
+func (m *Manager) prepareVolumes(ctx context.Context, svc config.ServiceConfig) ([]volume.PreparedVolume, error) {
+ if len(svc.Volumes) == 0 {
+ return nil, nil
+ }
+ if m.volumeManager == nil {
+ err := fmt.Errorf("service %s declares volumes but agent storage is not configured", svc.Name)
+ m.setVolumeError(svc.Name, err)
+ return nil, err
+ }
+ prepared, err := m.volumeManager.Prepare(ctx, svc)
+ if err != nil {
+ m.setVolumeError(svc.Name, err)
+ return nil, fmt.Errorf("preparing volumes: %w", err)
+ }
+ return prepared, nil
+}
+
+// discardStart removes this attempt's placeholder after a phase-2 failure. It
+// leaves a placeholder belonging to some later attempt alone.
+func (m *Manager) discardStart(name, startID string) {
+ m.mu.Lock()
+ defer m.mu.Unlock()
+ if inst, exists := m.instances[name]; exists && inst.startID == startID {
+ delete(m.instances, name)
+ }
+}
+
+// finishStart is phase 3: it re-takes the manager lock, confirms this attempt
+// still owns the placeholder, and only then writes durable state or launches.
+func (m *Manager) finishStart(ctx context.Context, svc config.ServiceConfig, startID, vmDir, socketPath string, prepared []volume.PreparedVolume) error {
+ m.mu.Lock()
+ defer m.mu.Unlock()
+
+ placeholder, exists := m.instances[svc.Name]
+ if !exists || placeholder.startID != startID || placeholder.State != StateStarting {
+ if exists && placeholder.startID == startID {
+ delete(m.instances, svc.Name)
}
+ m.logger.Info("start aborted before launch", "service", svc.Name)
+ return fmt.Errorf("service %s: %w", svc.Name, ErrStartAborted)
}
+ // Clamp the service configuration to what was actually prepared. A refused
+ // shrink prepares successfully at the applied size, and everything
+ // downstream — the config hash, the Firecracker config, the ownership
+ // manifest, and the instance the next tick compares against — must describe
+ // that effective configuration. Storing it here is what makes needsUpdate
+ // compare equal on the following tick rather than one convergence cycle
+ // later.
+ svc = clampToPrepared(svc, prepared)
+
configPath, err := m.writeVMConfig(vmDir, svc, prepared)
if err != nil {
+ delete(m.instances, svc.Name)
return fmt.Errorf("writing vm config: %w", err)
}
configHash, err := serviceConfigHash(svc)
if err != nil {
+ delete(m.instances, svc.Name)
return err
}
instanceID, err := newInstanceID()
if err != nil {
+ delete(m.instances, svc.Name)
return err
}
launcherKind, launcherUnit := startingLauncherMetadata(m.launcher, instanceID)
@@ -262,6 +477,7 @@ func (m *Manager) Start(ctx context.Context, svc config.ServiceConfig) error {
StartedAt: time.Now().UTC(), Volumes: append([]volume.PreparedVolume(nil), prepared...),
}
if err := writeManifest(manifestPath(vmDir), manifest); err != nil {
+ delete(m.instances, svc.Name)
return err
}
launched, err := m.launcher.Launch(ctx, launchSpec{
@@ -272,6 +488,7 @@ func (m *Manager) Start(ctx context.Context, svc config.ServiceConfig) error {
manifest.Lifecycle = lifecycleFailed
manifest.LastError = err.Error()
_ = writeManifest(manifestPath(vmDir), manifest)
+ delete(m.instances, svc.Name)
return fmt.Errorf("starting firecracker: %w", err)
}
manifest.PID = launched.PID
@@ -280,12 +497,15 @@ func (m *Manager) Start(ctx context.Context, svc config.ServiceConfig) error {
if identityErr := m.recordLaunchedIdentity(manifest, launched); identityErr != nil {
if m.abandonLaunch(svc.Name, manifest, launched, identityErr) {
m.instances[svc.Name] = instanceFromManifest(manifest, StateRecoveryPending, manifest.LastError)
+ } else {
+ delete(m.instances, svc.Name)
}
return fmt.Errorf("confirming launched process identity: %w", identityErr)
}
manifest.Lifecycle = lifecycleRunning
if err := writeManifest(manifestPath(vmDir), manifest); err != nil {
_ = m.launcher.Stop(manifest, syscall.SIGKILL)
+ delete(m.instances, svc.Name)
return err
}
@@ -298,6 +518,7 @@ func (m *Manager) Start(ctx context.Context, svc config.ServiceConfig) error {
Volumes: append([]volume.PreparedVolume(nil), prepared...),
instanceID: instanceID,
manifest: manifest,
+ startID: startID,
}
delete(m.volumeErrors, svc.Name)
@@ -475,6 +696,18 @@ func (m *Manager) Stop(name string) error {
m.mu.Unlock()
return err
}
+ // A start that is still preparing volumes has launched nothing, so there
+ // is no process to signal. Mark the attempt aborted and return
+ // immediately rather than waiting for a possibly multi-minute mkfs — not
+ // stalling shutdown behind volume work is the point of the barrier.
+ // Marking is idempotent so shutdown and the agent API can both issue a
+ // stop without the loser seeing an error for work the winner already did.
+ if inst.State == StateStarting || inst.State == StateStartAborting {
+ inst.State = StateStartAborting
+ m.mu.Unlock()
+ m.logger.Info("aborting in-flight start instead of stopping", "service", name)
+ return nil
+ }
manifest := inst.manifest
if manifest == nil {
m.mu.Unlock()
@@ -547,8 +780,25 @@ func (m *Manager) Stop(name string) error {
func (m *Manager) Remove(name string) error {
m.mu.Lock()
inst, exists := m.instances[name]
+ aborting := exists && (inst.State == StateStarting || inst.State == StateStartAborting)
+ if aborting {
+ inst.State = StateStartAborting
+ }
m.mu.Unlock()
+ // Removing the VM state directory while phase 2 runs is safe: volume
+ // preparation writes only under the storage pool, and writeVMConfig — the
+ // only writer of this directory — lives in phase 3, which will abort. The
+ // placeholder is left for phase 3 to clear, so a second Remove before then
+ // takes this same branch and also succeeds.
+ if aborting {
+ m.logger.Info("aborting in-flight start instead of removing", "service", name)
+ if err := os.RemoveAll(filepath.Join(m.stateDir, "vms", name)); err != nil {
+ return fmt.Errorf("removing vm dir: %w", err)
+ }
+ return nil
+ }
+
if exists && (inst.State == StateRunning || inst.State == StateStopping) {
if err := m.Stop(name); err != nil {
return fmt.Errorf("stopping VM during remove: %w", err)
@@ -664,11 +914,6 @@ func (m *Manager) quarantine(name string, manifest *instanceManifest, err error)
// writeVMConfig writes a Firecracker JSON config file for the given service.
func (m *Manager) writeVMConfig(vmDir string, svc config.ServiceConfig, prepared []volume.PreparedVolume) (string, error) {
- kernelArgs := svc.KernelArgs
- if kernelArgs == "" {
- kernelArgs = "console=ttyS0 reboot=k panic=1 pci=off"
- }
-
sort.Slice(prepared, func(i, j int) bool { return prepared[i].LogicalID < prepared[j].LogicalID })
drives := []firecrackerDrive{{DriveID: "rootfs", PathOnHost: svc.Image, IsRootDevice: true, IsReadOnly: false}}
guestVolumes := make([]guestVolume, 0, len(prepared))
@@ -686,13 +931,11 @@ func (m *Manager) writeVMConfig(vmDir string, svc config.ServiceConfig, prepared
MountPath: preparedVolume.MountPath, Type: preparedVolume.Type,
})
}
- if len(guestVolumes) > 0 {
- payload, err := json.Marshal(guestVolumePayload{Version: 1, Volumes: guestVolumes})
- if err != nil {
- return "", fmt.Errorf("marshal guest volume payload: %w", err)
- }
- arg := "firework.volumes64=" + base64.RawURLEncoding.EncodeToString(payload)
- kernelArgs = insertBeforeApplicationSeparator(kernelArgs, arg)
+ // The same builder Preflight uses, so an over-long command line now fails
+ // on create too rather than only on update.
+ kernelArgs, err := buildBootArgs(svc, guestVolumes)
+ if err != nil {
+ return "", err
}
var networkInterfaces []firecrackerNetworkInterface
diff --git a/internal/vm/manager_test.go b/internal/vm/manager_test.go
index 082241c..1197809 100644
--- a/internal/vm/manager_test.go
+++ b/internal/vm/manager_test.go
@@ -411,7 +411,7 @@ func TestPreflightRetainsVisibleVolumeError(t *testing.T) {
service := config.ServiceConfig{Name: "app", Volumes: []config.VolumeConfig{{
Name: "data", Type: config.VolumeTypeLocal, MountPath: "/data",
}}}
- if err := manager.Preflight(context.Background(), service); err == nil {
+ if _, err := manager.Preflight(context.Background(), service); err == nil {
t.Fatal("expected missing storage error")
}
if got := manager.VolumeError("app"); !strings.Contains(got, "storage is not configured") {
diff --git a/internal/vm/start_barrier_test.go b/internal/vm/start_barrier_test.go
new file mode 100644
index 0000000..a79a979
--- /dev/null
+++ b/internal/vm/start_barrier_test.go
@@ -0,0 +1,412 @@
+package vm
+
+import (
+ "context"
+ "errors"
+ "io"
+ "log/slog"
+ "os"
+ "path/filepath"
+ "strings"
+ "sync"
+ "syscall"
+ "testing"
+ "time"
+
+ "github.com/artemnikitin/firework/internal/config"
+ "github.com/artemnikitin/firework/internal/volume"
+)
+
+// blockingRunner parks inside the first filesystem-mutating command, which is
+// what a multi-minute mkfs.ext4 or resize2fs looks like from the manager's
+// point of view.
+type blockingRunner struct {
+ entered chan struct{}
+ release chan struct{}
+ once sync.Once
+ fail error
+}
+
+func (r *blockingRunner) Run(_ context.Context, name string, args ...string) ([]byte, error) {
+ if name == "resize2fs" && len(args) > 0 && args[0] == "-P" {
+ return []byte("Estimated minimum size of the filesystem: 1024\n"), nil
+ }
+ if name == "tune2fs" {
+ return []byte("Block size: 4096\n"), nil
+ }
+ return nil, nil
+}
+
+func (r *blockingRunner) RunDestructive(ctx context.Context, name string, args ...string) ([]byte, error) {
+ r.once.Do(func() {
+ close(r.entered)
+ <-r.release
+ })
+ if r.fail != nil {
+ return nil, r.fail
+ }
+ return r.Run(ctx, name, args...)
+}
+
+type acceptingMounts struct{}
+
+func (acceptingMounts) Verify(string) error { return nil }
+
+// fakeRunner reports a fixed filesystem minimum, so a small shrink target is
+// refused and a large one is accepted.
+type fakeRunner struct{}
+
+func (fakeRunner) Run(_ context.Context, name string, args ...string) ([]byte, error) {
+ if name == "resize2fs" && len(args) > 0 && args[0] == "-P" {
+ return []byte("Estimated minimum size of the filesystem: 1024\n"), nil
+ }
+ if name == "tune2fs" {
+ return []byte("Block size: 4096\n"), nil
+ }
+ return nil, nil
+}
+
+func (r fakeRunner) RunDestructive(ctx context.Context, name string, args ...string) ([]byte, error) {
+ return r.Run(ctx, name, args...)
+}
+
+// countingLauncher fails the test's purpose loudly: phase 3 must not launch
+// anything after an abort, so any launch at all is the defect.
+type countingLauncher struct {
+ mu sync.Mutex
+ launches int
+}
+
+func (l *countingLauncher) Launch(context.Context, launchSpec) (*launchedProcess, error) {
+ l.mu.Lock()
+ l.launches++
+ l.mu.Unlock()
+ return nil, errors.New("launch should not have been reached")
+}
+
+func (l *countingLauncher) Stop(*instanceManifest, syscall.Signal) error { return nil }
+
+func (l *countingLauncher) count() int {
+ l.mu.Lock()
+ defer l.mu.Unlock()
+ return l.launches
+}
+
+func barrierManager(t *testing.T, runner volume.CommandRunner) (*Manager, *countingLauncher) {
+ t.Helper()
+ stateDir := t.TempDir()
+ pool := t.TempDir()
+ volumeMgr := volume.NewManagerWithDependencies("node-1", config.StorageConfig{
+ Local: &config.LocalStorageConfig{Path: pool, CapacityBytes: 1 << 30},
+ }, runner, acceptingMounts{})
+ manager := NewManagerWithVolumes("/bin/true", stateDir, slog.New(slog.NewTextHandler(io.Discard, nil)), volumeMgr)
+ launcher := &countingLauncher{}
+ manager.launcher = launcher
+ return manager, launcher
+}
+
+func barrierService() config.ServiceConfig {
+ return config.ServiceConfig{
+ Name: "app", Image: "/image", Kernel: "/kernel", VCPUs: 1, MemoryMB: 128,
+ Volumes: []config.VolumeConfig{{
+ Name: "data", Type: config.VolumeTypeLocal, MountPath: "/var/lib/app",
+ SizeBytes: 16 * config.MiB, BoundNode: "node-1", ResizeGeneration: 1,
+ }},
+ }
+}
+
+// The whole point of releasing the lock: a reader must not block behind a
+// multi-minute volume operation, and what it reads must be truthful.
+func TestListReportsStartingWhilePrepareRuns(t *testing.T) {
+ runner := &blockingRunner{entered: make(chan struct{}), release: make(chan struct{})}
+ manager, launcher := barrierManager(t, runner)
+
+ done := make(chan error, 1)
+ go func() { done <- manager.Start(context.Background(), barrierService()) }()
+ <-runner.entered
+
+ instance := manager.List()["app"]
+ if instance == nil || instance.State != StateStarting {
+ t.Fatalf("expected a starting placeholder during Prepare, got %#v", instance)
+ }
+
+ close(runner.release)
+ <-done
+ if launcher.count() == 0 {
+ t.Fatal("expected the start to proceed to launch after Prepare returned")
+ }
+}
+
+// A stop that arrives while volumes are being prepared must return at once —
+// not wait out the mkfs — and the start must then launch nothing.
+func TestStopDuringPrepareAbortsTheStart(t *testing.T) {
+ runner := &blockingRunner{entered: make(chan struct{}), release: make(chan struct{})}
+ manager, launcher := barrierManager(t, runner)
+
+ done := make(chan error, 1)
+ go func() { done <- manager.Start(context.Background(), barrierService()) }()
+ <-runner.entered
+
+ stopped := make(chan error, 1)
+ go func() { stopped <- manager.Stop("app") }()
+ select {
+ case err := <-stopped:
+ if err != nil {
+ t.Fatalf("stopping a starting service should succeed, got %v", err)
+ }
+ case <-time.After(5 * time.Second):
+ t.Fatal("Stop blocked behind volume preparation")
+ }
+ // A second stop must not see an error for work the first one already did.
+ if err := manager.Stop("app"); err != nil {
+ t.Fatalf("repeated stop while aborting should succeed, got %v", err)
+ }
+
+ close(runner.release)
+ err := <-done
+ if !errors.Is(err, ErrStartAborted) {
+ t.Fatalf("expected ErrStartAborted, got %v", err)
+ }
+ if launcher.count() != 0 {
+ t.Fatalf("aborted start launched %d process(es)", launcher.count())
+ }
+ if instance := manager.List()["app"]; instance != nil {
+ t.Fatalf("expected the placeholder to be cleaned up, got %#v", instance)
+ }
+}
+
+func TestRemoveDuringPrepareAbortsTheStartAndClearsState(t *testing.T) {
+ runner := &blockingRunner{entered: make(chan struct{}), release: make(chan struct{})}
+ manager, launcher := barrierManager(t, runner)
+ vmDir := filepath.Join(manager.stateDir, "vms", "app")
+
+ done := make(chan error, 1)
+ go func() { done <- manager.Start(context.Background(), barrierService()) }()
+ <-runner.entered
+
+ removed := make(chan error, 1)
+ go func() { removed <- manager.Remove("app") }()
+ select {
+ case err := <-removed:
+ if err != nil {
+ t.Fatalf("removing a starting service should succeed, got %v", err)
+ }
+ case <-time.After(5 * time.Second):
+ t.Fatal("Remove blocked behind volume preparation")
+ }
+ if err := manager.Remove("app"); err != nil {
+ t.Fatalf("repeated remove while aborting should succeed, got %v", err)
+ }
+ if _, err := os.Stat(vmDir); !os.IsNotExist(err) {
+ t.Fatalf("expected the VM state directory to be removed, got %v", err)
+ }
+
+ close(runner.release)
+ if err := <-done; !errors.Is(err, ErrStartAborted) {
+ t.Fatalf("expected ErrStartAborted, got %v", err)
+ }
+ if launcher.count() != 0 {
+ t.Fatalf("aborted start launched %d process(es)", launcher.count())
+ }
+ if instance := manager.List()["app"]; instance != nil {
+ t.Fatalf("expected the placeholder to be cleaned up, got %#v", instance)
+ }
+}
+
+// Phase 3 validates the startID, not the service name, so a placeholder that
+// was cleared and replaced by a later attempt is never mistaken for its own.
+func TestPhaseThreeIgnoresAPlaceholderFromAnotherAttempt(t *testing.T) {
+ runner := &blockingRunner{entered: make(chan struct{}), release: make(chan struct{})}
+ manager, launcher := barrierManager(t, runner)
+
+ done := make(chan error, 1)
+ go func() { done <- manager.Start(context.Background(), barrierService()) }()
+ <-runner.entered
+
+ // Simulate the placeholder being taken over by a later start.
+ manager.mu.Lock()
+ manager.instances["app"] = &Instance{Name: "app", State: StateStarting, startID: "someone-else"}
+ manager.mu.Unlock()
+
+ close(runner.release)
+ if err := <-done; !errors.Is(err, ErrStartAborted) {
+ t.Fatalf("expected ErrStartAborted, got %v", err)
+ }
+ if launcher.count() != 0 {
+ t.Fatalf("start touched the launcher despite losing its placeholder (%d launches)", launcher.count())
+ }
+ if instance := manager.List()["app"]; instance == nil || instance.startID != "someone-else" {
+ t.Fatalf("expected the other attempt's placeholder to survive, got %#v", instance)
+ }
+}
+
+func TestFailedPrepareLeavesNoPlaceholder(t *testing.T) {
+ runner := &blockingRunner{entered: make(chan struct{}), release: make(chan struct{}), fail: errors.New("mkfs failed")}
+ manager, launcher := barrierManager(t, runner)
+
+ done := make(chan error, 1)
+ go func() { done <- manager.Start(context.Background(), barrierService()) }()
+ <-runner.entered
+ close(runner.release)
+
+ err := <-done
+ if err == nil || !strings.Contains(err.Error(), "preparing volumes") {
+ t.Fatalf("expected a volume preparation failure, got %v", err)
+ }
+ if instance := manager.List()["app"]; instance != nil {
+ t.Fatalf("expected no placeholder after a failed Prepare, got %#v", instance)
+ }
+ if launcher.count() != 0 {
+ t.Fatalf("failed Prepare still launched %d process(es)", launcher.count())
+ }
+ if manager.VolumeError("app") == "" {
+ t.Fatal("expected the preparation failure to stay visible as a volume error")
+ }
+}
+
+// The phase-1 entry check treats both starting states as active, so the agent
+// API and the shutdown path cannot start a second VM for the same service.
+func TestConcurrentStartIsRejectedWhileStarting(t *testing.T) {
+ runner := &blockingRunner{entered: make(chan struct{}), release: make(chan struct{})}
+ manager, _ := barrierManager(t, runner)
+
+ done := make(chan error, 1)
+ go func() { done <- manager.Start(context.Background(), barrierService()) }()
+ <-runner.entered
+
+ second := manager.Start(context.Background(), barrierService())
+ if !errors.Is(second, ErrStartInProgress) {
+ t.Fatalf("expected ErrStartInProgress, got %v", second)
+ }
+ if !IsStartRace(second) {
+ t.Fatal("a rejected concurrent start must classify as a benign race")
+ }
+
+ close(runner.release)
+ <-done
+}
+
+// An over-long command line must fail on create, not only on update: the
+// launch path and Preflight now build the args through the same function.
+func TestOverlongKernelCommandLineFailsOnCreate(t *testing.T) {
+ manager, _ := barrierManager(t, &fakeVolumeRunner{})
+ svc := barrierService()
+ svc.KernelArgs = strings.Repeat("x", maxKernelCommandLineBytes)
+
+ _, err := manager.writeVMConfig(t.TempDir(), svc, []volume.PreparedVolume{{
+ LogicalID: "app/data", PathOnHost: "/pool/app/data/volume.ext4",
+ MountPath: "/var/lib/app", Type: config.VolumeTypeLocal, SizeBytes: 16 * config.MiB,
+ }})
+ if err == nil || !strings.Contains(err.Error(), "kernel command line") {
+ t.Fatalf("expected the create path to enforce the command-line limit, got %v", err)
+ }
+}
+
+// Preflight rejects before a running VM is touched, and the launch path
+// rejects before it boots something unbootable. They must agree exactly.
+func TestPreflightAndWriteVMConfigBuildIdenticalBootArgs(t *testing.T) {
+ svc := barrierService()
+ svc.Volumes = append(svc.Volumes, config.VolumeConfig{
+ Name: "cache", Type: config.VolumeTypeLocal, MountPath: "/var/cache/app",
+ SizeBytes: 8 * config.MiB, BoundNode: "node-1", ResizeGeneration: 1,
+ })
+
+ declared, err := guestVolumesFromConfig(svc.Volumes)
+ if err != nil {
+ t.Fatal(err)
+ }
+ fromConfig, err := buildBootArgs(svc, declared)
+ if err != nil {
+ t.Fatal(err)
+ }
+
+ manager, _ := barrierManager(t, &fakeVolumeRunner{})
+ vmDir := t.TempDir()
+ prepared := []volume.PreparedVolume{
+ {LogicalID: "app/data", PathOnHost: "/pool/app/data/volume.ext4", MountPath: "/var/lib/app", Type: config.VolumeTypeLocal},
+ {LogicalID: "app/cache", PathOnHost: "/pool/app/cache/volume.ext4", MountPath: "/var/cache/app", Type: config.VolumeTypeLocal},
+ }
+ if _, err := manager.writeVMConfig(vmDir, svc, prepared); err != nil {
+ t.Fatal(err)
+ }
+ written, err := os.ReadFile(filepath.Join(vmDir, "vm-config.json"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ if !strings.Contains(string(written), jsonEscape(fromConfig)) {
+ t.Fatalf("preflight and launch boot args differ.\npreflight: %s\nwritten: %s", fromConfig, written)
+ }
+}
+
+// jsonEscape renders a boot-args string the way it appears inside the
+// Firecracker config document.
+func jsonEscape(value string) string {
+ return strings.ReplaceAll(value, `"`, `\"`)
+}
+
+// fakeVolumeRunner satisfies the runner interface for tests that never reach a
+// real filesystem operation.
+type fakeVolumeRunner struct{}
+
+func (fakeVolumeRunner) Run(context.Context, string, ...string) ([]byte, error) { return nil, nil }
+func (fakeVolumeRunner) RunDestructive(context.Context, string, ...string) ([]byte, error) {
+ return nil, nil
+}
+
+// The acknowledged form of a refused shrink must converge.
+//
+// After the control plane acknowledges a rejection it renders the *effective*
+// size with the *refused* generation — it keeps that generation so the
+// acknowledgement can match its own record. The running instance, meanwhile,
+// carries the applied generation. needsUpdate compares whole VolumeConfig
+// structs, so unless normalization reconciles the two the reconciler plans an
+// update, stops the VM, and restarts it — every tick that reaches Plan.
+func TestAcknowledgedRejectionConvergesWithTheRunningConfig(t *testing.T) {
+ manager, _ := barrierManager(t, &fakeRunner{})
+ svc := barrierService()
+
+ if _, err := manager.volumeManager.Prepare(context.Background(), svc); err != nil {
+ t.Fatal(err)
+ }
+ // A shrink the fake measurement refuses, at generation 2.
+ refused := svc
+ refused.Volumes = append([]config.VolumeConfig(nil), svc.Volumes...)
+ refused.Volumes[0].SizeBytes = 2 * config.MiB
+ refused.Volumes[0].ResizeGeneration = 2
+ prepared, err := manager.volumeManager.Prepare(context.Background(), refused)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if !prepared[0].Rejected {
+ t.Fatalf("precondition: expected the shrink to be refused, got %#v", prepared[0])
+ }
+
+ // What Start stores on the instance: the effective configuration.
+ running := clampToPrepared(refused, prepared)
+
+ // What the control plane renders once it has acknowledged the rejection.
+ acknowledged := svc
+ acknowledged.Volumes = append([]config.VolumeConfig(nil), svc.Volumes...)
+ acknowledged.Volumes[0].SizeBytes = 16 * config.MiB
+ acknowledged.Volumes[0].ResizeGeneration = 2
+
+ services := []config.ServiceConfig{acknowledged}
+ manager.NormalizeVolumes(services)
+
+ // volumesEqual compares whole VolumeConfig structs, so this equality is
+ // exactly the condition under which no ActionUpdate is planned.
+ if services[0].Volumes[0] != running.Volumes[0] {
+ t.Fatalf("the rendered config does not match the running one, so an update would be re-planned:\nrendered %#v\nrunning %#v",
+ services[0].Volumes[0], running.Volumes[0])
+ }
+
+ // The agent stops reporting a refusal here, and that is deliberate. These
+ // bytes are exactly what a direct-Git operator writes to *withdraw* the
+ // request, so the agent cannot tell a standing request from a withdrawn
+ // one and must not degrade the node forever on the ambiguity. Only the
+ // record still knows the operator's request, so that half of the
+ // visibility is the control plane's — see refusedVolumes there.
+ if got := manager.VolumeRejections(); len(got) != 0 {
+ t.Fatalf("the acknowledged shape must not keep reporting a refusal: %#v", got)
+ }
+}
diff --git a/internal/volume/hardening_test.go b/internal/volume/hardening_test.go
new file mode 100644
index 0000000..56c1787
--- /dev/null
+++ b/internal/volume/hardening_test.go
@@ -0,0 +1,274 @@
+package volume
+
+import (
+ "context"
+ "errors"
+ "os"
+ "path/filepath"
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/artemnikitin/firework/internal/config"
+)
+
+func hardeningManager(t *testing.T, runner CommandRunner) (*Manager, string) {
+ t.Helper()
+ root := t.TempDir()
+ return NewManagerWithDependencies("node-1", config.StorageConfig{Local: &config.LocalStorageConfig{
+ Path: root, CapacityBytes: 100 * config.MiB,
+ }}, runner, acceptingMounts{}), root
+}
+
+func volumePaths(root string) (dir, image, manifest, marker string) {
+ dir = filepath.Join(root, "app", "data")
+ return dir, filepath.Join(dir, imageFilename), filepath.Join(dir, manifestFilename), filepath.Join(dir, creationMarkerFilename)
+}
+
+// crashingRunner stops the create sequence at a chosen command, standing in for
+// a process that died partway through a first creation.
+type crashingRunner struct {
+ fakeRunner
+ failOn string
+}
+
+func (r *crashingRunner) RunDestructive(ctx context.Context, name string, args ...string) ([]byte, error) {
+ if name == r.failOn {
+ return nil, errors.New("simulated crash")
+ }
+ return r.fakeRunner.RunDestructive(ctx, name, args...)
+}
+
+// A crash after the image is created but before the manifest is written leaves
+// an empty volume. Nothing is protected by failing closed there, so the marker
+// makes it recoverable without an operator rm.
+func TestInterruptedCreationRecoversWithoutOperatorIntervention(t *testing.T) {
+ runner := &crashingRunner{failOn: "mkfs.ext4"}
+ manager, root := hardeningManager(t, runner)
+ dir, image, manifest, marker := volumePaths(root)
+
+ if _, err := manager.Prepare(context.Background(), localService(16*config.MiB, 1)); err == nil {
+ t.Fatal("expected the simulated crash to fail the first creation")
+ }
+ if _, err := os.Stat(image); err != nil {
+ t.Fatalf("expected the partially created image to remain: %v", err)
+ }
+ if _, err := os.Stat(marker); err != nil {
+ t.Fatalf("expected a creation marker to record the interrupted attempt: %v", err)
+ }
+
+ manager, _ = hardeningManager(t, &fakeRunner{})
+ manager.storage.Local.Path = root
+ prepared, err := manager.Prepare(context.Background(), localService(16*config.MiB, 1))
+ if err != nil {
+ t.Fatalf("expected the interrupted creation to be recoverable: %v", err)
+ }
+ if len(prepared) != 1 || prepared[0].SizeBytes != 16*config.MiB {
+ t.Fatalf("unexpected recovered volume: %#v", prepared)
+ }
+ if _, err := os.Stat(manifest); err != nil {
+ t.Fatalf("expected a manifest after recovery: %v", err)
+ }
+ if _, err := os.Stat(marker); !os.IsNotExist(err) {
+ t.Fatalf("expected the marker to be removed after a successful creation: %v", err)
+ }
+ _ = dir
+}
+
+// A crash between the image being sized and mkfs running is the same case one
+// step earlier, and must recover the same way.
+func TestInterruptedCreationBeforeMkfsRecovers(t *testing.T) {
+ manager, root := hardeningManager(t, &fakeRunner{})
+ dir, image, _, _ := volumePaths(root)
+ if err := os.MkdirAll(dir, 0o750); err != nil {
+ t.Fatal(err)
+ }
+ if err := writeCreationMarker(dir, "app", localService(16*config.MiB, 1).Volumes[0], "node-1"); err != nil {
+ t.Fatal(err)
+ }
+ if err := createSparseImage(image, 16*config.MiB); err != nil {
+ t.Fatal(err)
+ }
+
+ if _, err := manager.Prepare(context.Background(), localService(16*config.MiB, 1)); err != nil {
+ t.Fatalf("expected recovery from a sized-but-unformatted image: %v", err)
+ }
+}
+
+// The manifest was written but the process died before the marker was removed.
+// This is ordinary reuse — and the marker must not survive it, because a later
+// manifest loss would otherwise make a populated image look like an
+// interrupted empty creation and authorize deleting it.
+func TestSurvivingMarkerIsClearedOnReuse(t *testing.T) {
+ manager, root := hardeningManager(t, &fakeRunner{})
+ dir, _, _, marker := volumePaths(root)
+
+ if _, err := manager.Prepare(context.Background(), localService(16*config.MiB, 1)); err != nil {
+ t.Fatal(err)
+ }
+ // Recreate the state a crash between the manifest write and the marker
+ // removal leaves behind.
+ if err := writeCreationMarker(dir, "app", localService(16*config.MiB, 1).Volumes[0], "node-1"); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := manager.Prepare(context.Background(), localService(16*config.MiB, 1)); err != nil {
+ t.Fatalf("reuse failed: %v", err)
+ }
+ if _, err := os.Stat(marker); !os.IsNotExist(err) {
+ t.Fatalf("expected reuse to clear a marker that outlived its condition: %v", err)
+ }
+}
+
+// An image Firework did not create is exactly what failing closed is for, and
+// a marker that names a different node or volume proves nothing about this one.
+func TestUnprovenImageStaysQuarantined(t *testing.T) {
+ tests := []struct {
+ name string
+ marker *creationMarker
+ }{
+ {name: "no marker"},
+ {name: "marker names another node", marker: &creationMarker{LogicalID: "app/data", NodeID: "node-2"}},
+ {name: "marker names another volume", marker: &creationMarker{LogicalID: "other/data", NodeID: "node-1"}},
+ }
+ for _, test := range tests {
+ t.Run(test.name, func(t *testing.T) {
+ manager, root := hardeningManager(t, &fakeRunner{})
+ dir, image, _, marker := volumePaths(root)
+ if err := os.MkdirAll(dir, 0o750); err != nil {
+ t.Fatal(err)
+ }
+ if err := createSparseImage(image, 16*config.MiB); err != nil {
+ t.Fatal(err)
+ }
+ if test.marker != nil {
+ test.marker.CreatedAt = time.Now().UTC()
+ if err := writeJSONAtomic(marker, *test.marker); err != nil {
+ t.Fatal(err)
+ }
+ }
+ _, err := manager.Prepare(context.Background(), localService(16*config.MiB, 1))
+ if err == nil || !strings.Contains(err.Error(), "quarantined") {
+ t.Fatalf("expected the volume to stay quarantined, got %v", err)
+ }
+ })
+ }
+}
+
+// Filesystem-mutating commands must not run on a context the agent's signal
+// handler cancels, because exec.CommandContext cancellation is SIGKILL.
+func TestDestructiveCommandsDoNotTakeTheCancellablePath(t *testing.T) {
+ runner := &fakeRunner{}
+ manager, _ := hardeningManager(t, runner)
+
+ if _, err := manager.Prepare(context.Background(), localService(16*config.MiB, 1)); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := manager.Prepare(context.Background(), localService(24*config.MiB, 2)); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := manager.Prepare(context.Background(), localService(20*config.MiB, 3)); err != nil {
+ t.Fatal(err)
+ }
+
+ destructive := strings.Join(runner.destructive, "\n")
+ for _, want := range []string{"mkfs.ext4", "e2fsck", "resize2fs"} {
+ if !strings.Contains(destructive, want) {
+ t.Fatalf("expected %s to run on the uncancellable path, got:\n%s", want, destructive)
+ }
+ }
+ // The shrink measurement is read-only and must stay promptly cancellable.
+ for _, call := range runner.destructive {
+ if strings.HasPrefix(call, "resize2fs -P") || strings.HasPrefix(call, "tune2fs") {
+ t.Fatalf("measurement command %q must not be detached from the caller's context", call)
+ }
+ }
+}
+
+// A cancelled parent context must not stop a destructive command, which is the
+// whole reason RunDestructive exists.
+func TestRunDestructiveSurvivesParentCancellation(t *testing.T) {
+ ctx, cancel := context.WithCancel(context.Background())
+ cancel()
+
+ if _, err := (execRunner{}).RunDestructive(ctx, "true"); err != nil {
+ t.Fatalf("destructive command was killed by the cancelled parent context: %v", err)
+ }
+ if _, err := (execRunner{}).Run(ctx, "true"); err == nil {
+ t.Fatal("expected a read-only command to remain promptly cancellable")
+ }
+}
+
+// A retained manifest carrying a non-positive applied size subtracts from the
+// pool's reserved total, admitting a volume the pool cannot hold.
+func TestMalformedRetainedSizeCannotBypassPoolCapacity(t *testing.T) {
+ manager, root := hardeningManager(t, &fakeRunner{})
+ // A neighbouring service's retained manifest with a negative applied size.
+ dir := filepath.Join(root, "other", "data")
+ if err := os.MkdirAll(dir, 0o750); err != nil {
+ t.Fatal(err)
+ }
+ if err := writeJSONAtomic(filepath.Join(dir, manifestFilename), manifest{
+ LogicalID: "other/data", Type: config.VolumeTypeLocal, BoundNode: "node-1",
+ Filesystem: "ext4", AppliedSizeBytes: -100 * config.MiB, ResizeGeneration: 1,
+ }); err != nil {
+ t.Fatal(err)
+ }
+
+ // The pool is 100 MiB; 150 MiB must not fit regardless of the bad record.
+ _, err := manager.Preflight(context.Background(), localService(150*config.MiB, 1))
+ if err == nil {
+ t.Fatal("a negative retained size let an oversized volume into the pool")
+ }
+ if !strings.Contains(err.Error(), "capacity") && !strings.Contains(err.Error(), "quarantined") {
+ t.Fatalf("expected a capacity or quarantine failure, got %v", err)
+ }
+}
+
+// The agent derives a filesystem path from the service name, so a name that is
+// not a safe path component fails at preflight. configcheck must reject it too.
+func TestServiceNameIsValidatedByTheExportedValidator(t *testing.T) {
+ nc := config.NodeConfig{Node: "node-1", Services: []config.ServiceConfig{{
+ Name: "bad/name", Image: "/i", Kernel: "/k", VCPUs: 1, MemoryMB: 128,
+ Volumes: []config.VolumeConfig{{
+ Name: "data", Type: config.VolumeTypeLocal, MountPath: "/var/lib/app",
+ SizeBytes: config.MiB, BoundNode: "node-1", ResizeGeneration: 1,
+ }},
+ }}}
+ if err := ValidateNodeVolumes(nc); err == nil {
+ t.Fatal("a service name that is not a safe path component must be rejected")
+ }
+}
+
+// readRetained keys the pool's reservation map by the manifest's *declared*
+// LogicalID rather than by its path. Two manifests claiming the same logical ID
+// therefore collapse to one entry, and the other volume's bytes vanish from the
+// reserved total — a capacity bypass from state the node already holds.
+func TestRetainedManifestCannotMaskAnotherReservation(t *testing.T) {
+ manager, root := hardeningManager(t, &fakeRunner{})
+
+ // Two real, separate retained volumes, each 40 MiB in a 100 MiB pool.
+ for _, svc := range []string{"alpha", "beta"} {
+ dir := filepath.Join(root, svc, "data")
+ if err := os.MkdirAll(dir, 0o750); err != nil {
+ t.Fatal(err)
+ }
+ // beta's manifest lies about its identity and claims alpha's.
+ logicalID := svc + "/data"
+ if svc == "beta" {
+ logicalID = "alpha/data"
+ }
+ if err := writeJSONAtomic(filepath.Join(dir, manifestFilename), manifest{
+ LogicalID: logicalID, Type: config.VolumeTypeLocal, BoundNode: "node-1",
+ Filesystem: "ext4", AppliedSizeBytes: 40 * config.MiB, ResizeGeneration: 1,
+ }); err != nil {
+ t.Fatal(err)
+ }
+ }
+
+ // 80 MiB is genuinely retained. A new 40 MiB volume would need 120 MiB of a
+ // 100 MiB pool and must be refused.
+ _, err := manager.Preflight(context.Background(), localService(40*config.MiB, 1))
+ if err == nil {
+ t.Fatal("a mislabelled retained manifest masked another volume's reservation")
+ }
+}
diff --git a/internal/volume/manager.go b/internal/volume/manager.go
index f140a11..c136485 100644
--- a/internal/volume/manager.go
+++ b/internal/volume/manager.go
@@ -15,6 +15,7 @@ import (
"sort"
"strconv"
"strings"
+ "sync"
"syscall"
"time"
@@ -25,6 +26,10 @@ const (
manifestFilename = "manifest.json"
transactionFilename = "resize-transaction.json"
imageFilename = "volume.ext4"
+ // creationMarkerFilename records that a first creation is in flight. It is
+ // what separates "we crashed while making an empty image" from "an image
+ // Firework did not create", which the manifest's absence alone cannot.
+ creationMarkerFilename = "creating.json"
)
var (
@@ -34,14 +39,68 @@ var (
ErrSharedUnsupported = errors.New("shared volumes require the durable per-VM supervisor and fencing validation")
)
+// ErrShrinkRejected reports that a requested shrink is below the safe minimum
+// for the filesystem's current contents. It is a *decision*, not a fault: the
+// distinction is what lets the caller keep the workload running instead of
+// treating a refusal like a failed operation.
+//
+// LogicalID is carried because two volumes sharing a size and generation are
+// otherwise indistinguishable, and the clamp cannot tell which one it applies
+// to.
+type ErrShrinkRejected struct {
+ LogicalID string
+ Requested int64
+ Minimum int64
+ Generation int64
+}
+
+func (e *ErrShrinkRejected) Error() string {
+ return fmt.Sprintf("volume %s: shrink target %d is below safe minimum %d", e.LogicalID, e.Requested, e.Minimum)
+}
+
+// Rejection is a durable refusal of one volume's size request, as reported to
+// status and consumed by the agent-side clamp.
+type Rejection struct {
+ LogicalID string
+ // ResizeGeneration is the *requested* generation — the one that was
+ // refused. It is what a reported rejection must carry so the control
+ // plane's acknowledgement can match it to the record it has to converge.
+ ResizeGeneration int64
+ // AppliedGeneration is the generation actually applied to the filesystem.
+ // Together with AppliedSizeBytes it is the *effective* configuration: what
+ // the node is running, what Plan compares against, and what the clamp
+ // substitutes. Keeping the two apart is what lets one rejection be both
+ // terminal locally and matchable remotely.
+ AppliedGeneration int64
+ RequestedSizeBytes int64
+ AppliedSizeBytes int64
+ MinimumSizeBytes int64
+ At time.Time
+}
+
// PreparedVolume is safe to attach to a stopped/new Firecracker process.
type PreparedVolume struct {
- LogicalID string
- PathOnHost string
- MountPath string
- Type config.VolumeType
- SizeBytes int64
+ LogicalID string
+ PathOnHost string
+ MountPath string
+ Type config.VolumeType
+ // SizeBytes is the *effective* size: what the image actually is. For a
+ // rejected shrink this is the applied size, not the refused request.
+ SizeBytes int64
+ // ResizeGeneration is always the generation actually applied to the
+ // filesystem. Together with SizeBytes it is the effective configuration
+ // the caller stores on the instance, which is what makes the next tick
+ // compare equal instead of re-planning the same update.
ResizeGeneration int64
+ // Rejected marks a preparation that succeeded at a size other than the one
+ // requested. It is not an error, so Prepare continues to the next volume
+ // and one pass collects every rejection.
+ Rejected bool
+ // RequestedGeneration and RequestedSizeBytes describe the refused request.
+ // They are reported rather than run.
+ RequestedGeneration int64
+ RequestedSizeBytes int64
+ MinimumSizeBytes int64
}
// Status is the agent-observed state of one logical volume.
@@ -64,7 +123,102 @@ type manifest struct {
Filesystem string `json:"filesystem"`
AppliedSizeBytes int64 `json:"applied_size_bytes"`
ResizeGeneration int64 `json:"resize_generation"`
- UpdatedAt time.Time `json:"updated_at"`
+ // The rejection is keyed to one (generation, size) request. Recording it
+ // durably is what makes the refusal terminal without depending on the
+ // control plane: the agent-side clamp reads it locally, so the stop/restart
+ // loop is broken even if the acknowledgement never lands.
+ //
+ // ResizeGeneration deliberately continues to describe the last generation
+ // actually applied to the filesystem. Advancing it here would erase the
+ // evidence that this generation was refused.
+ RejectedGeneration int64 `json:"rejected_generation,omitempty"`
+ RejectedSizeBytes int64 `json:"rejected_size_bytes,omitempty"`
+ RejectedMinimumBytes int64 `json:"rejected_minimum_bytes,omitempty"`
+ RejectedAt time.Time `json:"rejected_at,omitempty"`
+ UpdatedAt time.Time `json:"updated_at"`
+}
+
+// rejectionFor builds the reported rejection for a manifest that carries one.
+func (m manifest) rejectionFor(generation int64) Rejection {
+ return Rejection{
+ LogicalID: m.LogicalID, ResizeGeneration: generation, AppliedGeneration: m.ResizeGeneration,
+ RequestedSizeBytes: m.RejectedSizeBytes, AppliedSizeBytes: m.AppliedSizeBytes,
+ MinimumSizeBytes: m.RejectedMinimumBytes, At: m.RejectedAt,
+ }
+}
+
+// matchesRejection reports whether a desired volume config is a request this
+// manifest already refused.
+//
+// The generation must always match. A generation-only match is not enough:
+// direct-Git node configs are hand-authored and carry their own
+// resize_generation, so an operator correcting a refused shrink by editing
+// size_bytes alone presents a different request under the same generation, and
+// a generation-only match would clamp that forever.
+//
+// Two sizes carry the same refused request, and both must be recognized:
+//
+// - the refused size itself, which is what a direct-Git config renders and
+// what the control plane renders until it has acknowledged the refusal;
+// - the applied size, which is what the control plane renders *after*
+// acknowledging it — the clamp there substitutes the effective size but
+// keeps the refused generation, because the acknowledgement has to be able
+// to match that generation to its record.
+//
+// Recognizing only the first leaves the second carrying the refused generation
+// while the running instance carries the applied one. needsUpdate compares
+// whole volume configs, so the service is stopped and restarted on every
+// reconcile that reaches Plan — the loop this whole mechanism exists to end.
+func (m manifest) matchesRejection(volume config.VolumeConfig) bool {
+ if m.RejectedGeneration == 0 || m.RejectedGeneration != volume.ResizeGeneration {
+ return false
+ }
+ return volume.SizeBytes == m.RejectedSizeBytes || volume.SizeBytes == m.AppliedSizeBytes
+}
+
+// refusesRequest reports whether the config in front of the agent is still
+// asking for the size that was refused.
+//
+// This is deliberately narrower than matchesRejection, and the two must not be
+// conflated. Clamping has to keep applying to both shapes for as long as the
+// refused generation stands, or the generation diverges from the running
+// instance and the service restarts on every reconcile. But a *report* of a
+// standing refusal is only true while the refused size is actually being
+// requested: once the config asks for the size already running — because a
+// direct-Git operator withdrew the request, or the control plane acknowledged
+// the refusal and now renders the effective size — nothing is being refused
+// here any more. Reporting one anyway leaves the node degraded forever with no
+// exit but a generation bump.
+//
+// After the control plane acknowledges, the two shapes become identical bytes
+// and the agent genuinely cannot tell whether the operator still wants the
+// refused size. Only the record knows, so that half of the visibility belongs
+// to the control plane; see §7.3.2 of the hardening plan.
+func (m manifest) refusesRequest(volume config.VolumeConfig) bool {
+ return m.RejectedGeneration != 0 &&
+ m.RejectedGeneration == volume.ResizeGeneration &&
+ m.RejectedSizeBytes == volume.SizeBytes
+}
+
+func (m *manifest) clearRejection() {
+ m.RejectedGeneration = 0
+ m.RejectedSizeBytes = 0
+ m.RejectedMinimumBytes = 0
+ m.RejectedAt = time.Time{}
+}
+
+// creationMarker is written before the backing image and removed after the
+// manifest. Its presence authorizes deleting an image that has no manifest, so
+// its lifetime is deliberately bounded by the condition it describes: every
+// path that reads a valid manifest removes a matching marker (see
+// clearStaleCreationMarker). A marker that outlived a successful creation would
+// otherwise authorize destroying populated data if the manifest were later lost.
+type creationMarker struct {
+ LogicalID string `json:"logical_id"`
+ NodeID string `json:"node_id"`
+ TargetSizeBytes int64 `json:"target_size_bytes"`
+ ResizeGeneration int64 `json:"resize_generation"`
+ CreatedAt time.Time `json:"created_at"`
}
type resizeTransaction struct {
@@ -76,15 +230,62 @@ type resizeTransaction struct {
UpdatedAt time.Time `json:"updated_at"`
}
+// destructiveCommandTimeout bounds a filesystem-mutating command that has been
+// detached from the caller's context. It has to accommodate mkfs, e2fsck, and
+// resize2fs on a pool-sized image, so it is generous: the point is that the
+// operation is not killed by an agent restart, not that it is killed promptly.
+//
+// Detaching from the Go context is only half of the protection, and the other
+// half is not in this repository. Under systemd's default
+// KillMode=control-group, stopping the agent's unit signals every process in
+// its cgroup — including this child — and force-kills the group at
+// TimeoutStopSec. The unit must set KillMode=mixed and a TimeoutStopSec above
+// this value; see docs/persistent-volumes.md. Raising this constant without
+// raising that one reopens the gap it exists to close.
+const destructiveCommandTimeout = 30 * time.Minute
+
+// destructiveCommandGrace is how long a timed-out destructive command is given
+// to handle SIGTERM before the process group is killed.
+const destructiveCommandGrace = 10 * time.Second
+
// CommandRunner isolates filesystem utilities for unit tests.
+//
+// The split between Run and RunDestructive is the interface's whole point, and
+// it lives here rather than in a name match inside the runner so a new
+// filesystem-mutating command cannot inherit the cancellable path by omission.
type CommandRunner interface {
+ // Run executes a read-only measurement command. It keeps the caller's
+ // context and stays promptly cancellable.
Run(context.Context, string, ...string) ([]byte, error)
+ // RunDestructive executes a command that mutates a filesystem. It must not
+ // be killed when the caller's context is cancelled: the agent's context is
+ // cancelled on SIGINT/SIGTERM, and exec.CommandContext cancellation is
+ // SIGKILL, so a systemd restart or node drain during a shrink would
+ // SIGKILL resize2fs mid-operation.
+ RunDestructive(context.Context, string, ...string) ([]byte, error)
}
type execRunner struct{}
func (execRunner) Run(ctx context.Context, name string, args ...string) ([]byte, error) {
- output, err := exec.CommandContext(ctx, name, args...).CombinedOutput()
+ return runCommand(exec.CommandContext(ctx, name, args...), name, args)
+}
+
+func (execRunner) RunDestructive(ctx context.Context, name string, args ...string) ([]byte, error) {
+ // WithoutCancel keeps the values (and therefore any tracing) from the
+ // caller's context while detaching it from the SIGTERM cancellation chain.
+ // The command then gets its own absolute deadline, and that deadline is a
+ // SIGTERM with a grace period rather than an unconditional SIGKILL.
+ detached, cancel := context.WithTimeout(context.WithoutCancel(ctx), destructiveCommandTimeout)
+ defer cancel()
+ cmd := exec.CommandContext(detached, name, args...)
+ cmd.Cancel = func() error { return cmd.Process.Signal(syscall.SIGTERM) }
+ cmd.WaitDelay = destructiveCommandGrace
+ return runCommand(cmd, name, args)
+}
+
+func runCommand(cmd *exec.Cmd, name string, args []string) ([]byte, error) {
+ output, err := cmd.CombinedOutput()
if err != nil {
return output, fmt.Errorf("%s %s: %w: %s", name, strings.Join(args, " "), err, strings.TrimSpace(string(output)))
}
@@ -139,10 +340,19 @@ type Manager struct {
runner CommandRunner
mounts MountVerifier
observer Observer
+
+ // rejections is the synchronized per-volume refusal snapshot, updated
+ // wherever a rejection is recorded — preflight or post-stop. Status reads
+ // it directly rather than inferring state from a running instance's
+ // prepared volumes, because a preflight rejection produces no fresh
+ // preparation to read: it fails the update before anything is stopped, so
+ // the instance still describes the *previous* preparation.
+ rejectionMu sync.RWMutex
+ rejections map[string]Rejection
}
func NewManager(nodeID string, storage config.StorageConfig) *Manager {
- return &Manager{nodeID: nodeID, storage: storage, runner: execRunner{}, mounts: procMountVerifier{}}
+ return &Manager{nodeID: nodeID, storage: storage, runner: execRunner{}, mounts: procMountVerifier{}, rejections: make(map[string]Rejection)}
}
func NewManagerWithObserver(nodeID string, storage config.StorageConfig, observer Observer) *Manager {
@@ -152,73 +362,145 @@ func NewManagerWithObserver(nodeID string, storage config.StorageConfig, observe
}
func NewManagerWithDependencies(nodeID string, storage config.StorageConfig, runner CommandRunner, mounts MountVerifier) *Manager {
- return &Manager{nodeID: nodeID, storage: storage, runner: runner, mounts: mounts}
+ return &Manager{nodeID: nodeID, storage: storage, runner: runner, mounts: mounts, rejections: make(map[string]Rejection)}
}
-// Preflight validates every declaration and retained image without mutating it.
-func (m *Manager) Preflight(ctx context.Context, svc config.ServiceConfig) error {
- _ = ctx
+// Preflight validates every declaration and retained image without mutating it,
+// apart from recording a refusal.
+//
+// It returns the rejections it found alongside its error rather than returning
+// on the first one. A rejection is a decision and a failure is a fault: only
+// the latter aborts the batch, so one pass collects every refusal and the
+// caller never has to retry per volume.
+func (m *Manager) Preflight(ctx context.Context, svc config.ServiceConfig) ([]Rejection, error) {
if len(svc.Volumes) == 0 {
- return nil
+ return nil, nil
}
if err := validateServiceVolumes(svc.Volumes); err != nil {
- return fmt.Errorf("service %s: %w", svc.Name, err)
+ return nil, fmt.Errorf("service %s: %w", svc.Name, err)
}
+ var rejections []Rejection
desiredLocal := make(map[string]int64)
for _, volume := range svc.Volumes {
logicalID := svc.Name + "/" + volume.Name
switch volume.Type {
case config.VolumeTypeLocal:
if m.storage.Local == nil {
- return fmt.Errorf("volume %s: storage.local is not configured", logicalID)
+ return rejections, fmt.Errorf("volume %s: storage.local is not configured", logicalID)
}
if volume.BoundNode == "" || volume.BoundNode != m.nodeID {
- return fmt.Errorf("volume %s: bound_node %q does not match node %q", logicalID, volume.BoundNode, m.nodeID)
+ return rejections, fmt.Errorf("volume %s: bound_node %q does not match node %q", logicalID, volume.BoundNode, m.nodeID)
}
if m.mounts != nil {
if err := m.mounts.Verify(m.storage.Local.Path); err != nil {
- return fmt.Errorf("volume %s: verify local storage: %w", logicalID, err)
+ return rejections, fmt.Errorf("volume %s: verify local storage: %w", logicalID, err)
}
}
desiredLocal[logicalID] = volume.SizeBytes
if err := m.validateExisting(svc.Name, volume, m.storage.Local.Path); err != nil {
- return err
+ return rejections, err
+ }
+ rejection, err := m.preflightResize(ctx, svc.Name, volume, m.storage.Local.Path)
+ if err != nil {
+ return rejections, err
}
- if err := m.preflightResize(ctx, svc.Name, volume, m.storage.Local.Path); err != nil {
- return err
+ if rejection != nil {
+ // The effective size is what capacity should be checked
+ // against; charging the refused request would reject a
+ // configuration the node is already running.
+ desiredLocal[logicalID] = rejection.AppliedSizeBytes
+ rejections = append(rejections, *rejection)
}
case config.VolumeTypeShared:
- return fmt.Errorf("volume %s: %w", logicalID, ErrSharedUnsupported)
+ return rejections, fmt.Errorf("volume %s: %w", logicalID, ErrSharedUnsupported)
default:
- return fmt.Errorf("volume %s: unsupported type %q", logicalID, volume.Type)
+ return rejections, fmt.Errorf("volume %s: unsupported type %q", logicalID, volume.Type)
}
}
if len(desiredLocal) > 0 {
if err := m.checkCapacity(m.storage.Local, desiredLocal); err != nil {
- return err
+ return rejections, err
}
}
- return nil
+ m.refreshRejections(svc)
+ return rejections, nil
}
-func (m *Manager) preflightResize(ctx context.Context, service string, volume config.VolumeConfig, root string) error {
+// preflightResize measures a requested shrink before anything is stopped, and
+// records a refusal durably so the refusal is terminal rather than re-measured
+// on every tick forever.
+//
+// The measurement is advisory: a live resize2fs -P errs in both directions,
+// because guest deletions whose bitmap updates are still in the page cache read
+// too large and guest writes not yet flushed read too small. Terminality is
+// still the right call, because the costs are asymmetric — a false refusal
+// costs the operator one re-request, which mints a new generation and
+// re-measures from scratch, while a non-terminal preflight costs an unbounded
+// measurement loop on every tick.
+//
+// The whole read → measure → write sequence runs under the volume's lifecycle
+// lock, the same lock prepareOne takes. Preflight used to be a pure reader; now
+// that it writes the manifest it can interleave with a concurrent Prepare, and
+// a measurement taken under one lock and written under another is the same lost
+// update with extra steps.
+func (m *Manager) preflightResize(ctx context.Context, service string, volume config.VolumeConfig, root string) (*Rejection, error) {
dir, err := volumeDir(root, service, volume.Name)
if err != nil {
- return err
+ return nil, err
+ }
+ manifestPath := filepath.Join(dir, manifestFilename)
+ if _, statErr := os.Stat(manifestPath); statErr != nil {
+ if os.IsNotExist(statErr) {
+ return nil, nil
+ }
+ return nil, statErr
+ }
+ lock, err := lockFile(filepath.Join(dir, "lifecycle.lock"))
+ if err != nil {
+ return nil, err
}
+ defer unlockFile(lock)
+
var current manifest
- if err := readJSON(filepath.Join(dir, manifestFilename), ¤t); err != nil {
+ if err := readJSON(manifestPath, ¤t); err != nil {
if os.IsNotExist(err) {
- return nil
+ return nil, nil
}
- return err
+ return nil, err
+ }
+ if current.refusesRequest(volume) {
+ // Already refused, for exactly this request. Re-measuring would be the
+ // unbounded loop this record exists to stop. A config that merely
+ // carries the refused *generation* at the effective size is not a
+ // refusal — nothing is being asked for that was denied — and falls
+ // through to the size comparison below, which finds nothing to do.
+ rejection := current.rejectionFor(volume.ResizeGeneration)
+ return &rejection, nil
}
if volume.SizeBytes >= current.AppliedSizeBytes {
- return nil
+ return nil, nil
+ }
+ err = m.inspectShrinkMinimum(ctx, service, volume, filepath.Join(dir, imageFilename))
+ var rejected *ErrShrinkRejected
+ if errors.As(err, &rejected) {
+ // Nothing has been stopped and no resize has begun, so there is no
+ // transaction to clean up here — only the manifest write applies.
+ current.RejectedGeneration = volume.ResizeGeneration
+ current.RejectedSizeBytes = volume.SizeBytes
+ current.RejectedMinimumBytes = rejected.Minimum
+ current.RejectedAt = time.Now().UTC()
+ current.UpdatedAt = current.RejectedAt
+ if writeErr := writeJSONAtomic(manifestPath, current); writeErr != nil {
+ return nil, writeErr
+ }
+ if syncErr := syncDir(dir); syncErr != nil {
+ return nil, syncErr
+ }
+ rejection := current.rejectionFor(volume.ResizeGeneration)
+ return &rejection, nil
}
- imagePath := filepath.Join(dir, imageFilename)
- return m.inspectShrinkMinimum(ctx, service, volume, imagePath)
+ return nil, err
}
func (m *Manager) inspectShrinkMinimum(ctx context.Context, service string, volume config.VolumeConfig, imagePath string) error {
@@ -243,7 +525,10 @@ func (m *Manager) inspectShrinkMinimum(ctx context.Context, service string, volu
// guarantee and can change after a final fsck.
minimumWithHeadroom := minimumBytes + minimumBytes/20
if volume.SizeBytes < minimumWithHeadroom {
- return fmt.Errorf("volume %s/%s: shrink target %d is below safe minimum %d", service, volume.Name, volume.SizeBytes, minimumWithHeadroom)
+ return &ErrShrinkRejected{
+ LogicalID: service + "/" + volume.Name, Requested: volume.SizeBytes,
+ Minimum: minimumWithHeadroom, Generation: volume.ResizeGeneration,
+ }
}
return nil
}
@@ -251,7 +536,7 @@ func (m *Manager) inspectShrinkMinimum(ctx context.Context, service string, volu
// Prepare creates/reuses/resizes all service images in deterministic order.
// Callers must invoke Preflight before stopping a running VM.
func (m *Manager) Prepare(ctx context.Context, svc config.ServiceConfig) ([]PreparedVolume, error) {
- if err := m.Preflight(ctx, svc); err != nil {
+ if _, err := m.Preflight(ctx, svc); err != nil {
if m.observer != nil {
outcome := "failure"
if strings.Contains(err.Error(), "quarantined") {
@@ -270,10 +555,14 @@ func (m *Manager) Prepare(ctx context.Context, svc config.ServiceConfig) ([]Prep
root := m.storage.Local.Path
p, err := m.prepareOne(ctx, svc.Name, volume, root)
if err != nil {
+ // A genuine failure still aborts the batch: a rejection is a
+ // decision, a failure is a fault, and only the latter means the
+ // remaining volumes cannot be trusted.
return nil, err
}
prepared = append(prepared, p)
}
+ m.refreshRejections(svc)
return prepared, nil
}
@@ -288,6 +577,16 @@ func (m *Manager) validateExisting(service string, volume config.VolumeConfig, r
if err := readJSON(manifestPath, &found); err != nil {
if os.IsNotExist(err) {
if _, statErr := os.Stat(imagePath); statErr == nil {
+ // An image with no manifest is either a creation this node
+ // crashed partway through — recoverable, because the image is
+ // empty and nothing is protected by failing closed — or an
+ // image Firework did not create, which is exactly what
+ // fail-closed exists for. Only a matching marker tells them
+ // apart, so an absent, unreadable, or mismatched marker still
+ // quarantines.
+ if matchingCreationMarker(dir, service, volume, m.nodeID) {
+ return nil
+ }
return fmt.Errorf("volume %s/%s: image exists without manifest; quarantined", service, volume.Name)
}
return nil
@@ -327,6 +626,7 @@ func (m *Manager) prepareOne(ctx context.Context, service string, volume config.
}
m.observer.ObserveVolumeOperation(string(volume.Type), operation, outcome, time.Since(started))
}()
+ var rejection *Rejection
dir, err := volumeDir(root, service, volume.Name)
if err != nil {
return PreparedVolume{}, err
@@ -346,16 +646,25 @@ func (m *Manager) prepareOne(ctx context.Context, service string, volume config.
err = readJSON(manifestPath, ¤t)
if os.IsNotExist(err) {
operation = "create"
+ if err := m.clearInterruptedCreation(dir, imagePath, service, volume); err != nil {
+ return PreparedVolume{}, err
+ }
+ if err := writeCreationMarker(dir, service, volume, m.nodeID); err != nil {
+ return PreparedVolume{}, err
+ }
if err := createSparseImage(imagePath, volume.SizeBytes); err != nil {
return PreparedVolume{}, err
}
- if _, err := m.runner.Run(ctx, "mkfs.ext4", "-F", "-m", "0", imagePath); err != nil {
+ if _, err := m.runner.RunDestructive(ctx, "mkfs.ext4", "-F", "-m", "0", imagePath); err != nil {
return PreparedVolume{}, err
}
current = manifestFor(service, volume, m.nodeID)
if err := writeJSONAtomic(manifestPath, current); err != nil {
return PreparedVolume{}, err
}
+ if err := removeCreationMarker(dir); err != nil {
+ return PreparedVolume{}, err
+ }
} else if err != nil {
return PreparedVolume{}, fmt.Errorf("read manifest: %w", err)
} else {
@@ -363,6 +672,13 @@ func (m *Manager) prepareOne(ctx context.Context, service string, volume config.
if err := verifyManifest(current, service, volume, m.nodeID); err != nil {
return PreparedVolume{}, err
}
+ // The manifest is valid, so any surviving marker describes a condition
+ // that has already ended — a crash between the manifest write and the
+ // marker removal. Clearing it here is what stops it from authorizing a
+ // delete later, if the manifest is ever lost.
+ if err := removeCreationMarker(dir); err != nil {
+ return PreparedVolume{}, err
+ }
transactionPath := filepath.Join(dir, transactionFilename)
var stale resizeTransaction
transactionErr := readJSON(transactionPath, &stale)
@@ -380,7 +696,27 @@ func (m *Manager) prepareOne(ctx context.Context, service string, volume config.
return PreparedVolume{}, err
}
}
- if current.AppliedSizeBytes != volume.SizeBytes || current.ResizeGeneration != volume.ResizeGeneration {
+ // The clamped configuration needs its own branch, evaluated before the
+ // resize condition. The clamp substitutes the applied size but keeps
+ // the *requested* generation — which is what the acknowledgement has
+ // to match — so it still satisfies the generation arm below and would
+ // re-enter resize forever. A short-circuit keyed on the requested
+ // (generation, size) pair cannot help either: the manifest records the
+ // rejection at the refused size while the clamped input presents the
+ // applied one, so the two never match by construction.
+ //
+ // The applied-size equality is what keeps a genuinely new request from
+ // being clamped: a raw config arriving at a matching generation but a
+ // non-applied size falls through to resize and re-measures.
+ if current.RejectedGeneration != 0 && current.RejectedGeneration == volume.ResizeGeneration &&
+ volume.SizeBytes == current.AppliedSizeBytes {
+ operation = "rejected"
+ rejection = &Rejection{
+ LogicalID: current.LogicalID, ResizeGeneration: volume.ResizeGeneration,
+ RequestedSizeBytes: current.RejectedSizeBytes, AppliedSizeBytes: current.AppliedSizeBytes,
+ MinimumSizeBytes: current.RejectedMinimumBytes, At: current.RejectedAt,
+ }
+ } else if current.AppliedSizeBytes != volume.SizeBytes || current.ResizeGeneration != volume.ResizeGeneration {
operation = "grow"
if volume.SizeBytes < current.AppliedSizeBytes {
operation = "shrink"
@@ -394,20 +730,39 @@ func (m *Manager) prepareOne(ctx context.Context, service string, volume config.
return PreparedVolume{}, fmt.Errorf("volume %s/%s: resize transaction does not match desired generation; quarantined", service, volume.Name)
}
}
- if err := m.resize(ctx, dir, imagePath, ¤t, volume); err != nil {
+ resized, err := m.resize(ctx, dir, imagePath, ¤t, volume)
+ if err != nil {
return PreparedVolume{}, err
}
+ rejection = resized
}
}
- return PreparedVolume{
+ prepared = PreparedVolume{
LogicalID: service + "/" + volume.Name, PathOnHost: imagePath,
MountPath: volume.MountPath, Type: volume.Type, SizeBytes: current.AppliedSizeBytes,
ResizeGeneration: current.ResizeGeneration,
- }, nil
+ }
+ if rejection != nil {
+ // A rejection is a non-fatal outcome of a *successful* preparation, so
+ // no error is returned and Prepare continues to the next volume. One
+ // pass therefore collects every rejection, without a retry budget that
+ // a second rejected volume would exhaust.
+ prepared.Rejected = true
+ prepared.RequestedSizeBytes = rejection.RequestedSizeBytes
+ prepared.RequestedGeneration = rejection.ResizeGeneration
+ prepared.MinimumSizeBytes = rejection.MinimumSizeBytes
+ }
+ return prepared, nil
}
-func (m *Manager) resize(ctx context.Context, dir, imagePath string, current *manifest, desired config.VolumeConfig) error {
+// resize applies a size change, or refuses one.
+//
+// A refusal returns a Rejection and no error, because by this point
+// deleteService has already stopped the VM: treating the refusal as a failure
+// would leave the workload down, which is precisely what the caller must be
+// able to avoid.
+func (m *Manager) resize(ctx context.Context, dir, imagePath string, current *manifest, desired config.VolumeConfig) (*Rejection, error) {
transactionPath := filepath.Join(dir, transactionFilename)
direction := "grow"
if desired.SizeBytes < current.AppliedSizeBytes {
@@ -418,63 +773,112 @@ func (m *Manager) resize(ctx context.Context, dir, imagePath string, current *ma
Generation: desired.ResizeGeneration, Direction: direction, Phase: "checking", UpdatedAt: time.Now().UTC(),
}
if err := writeJSONAtomic(transactionPath, tx); err != nil {
- return fmt.Errorf("write resize transaction: %w", err)
+ return nil, fmt.Errorf("write resize transaction: %w", err)
}
- if _, err := m.runner.Run(ctx, "e2fsck", "-f", "-y", imagePath); err != nil {
- return err
+ if _, err := m.runner.RunDestructive(ctx, "e2fsck", "-f", "-y", imagePath); err != nil {
+ return nil, err
}
if direction == "shrink" {
parts := strings.SplitN(current.LogicalID, "/", 2)
service := parts[0]
- if err := m.inspectShrinkMinimum(ctx, service, desired, imagePath); err != nil {
- return err
+ err := m.inspectShrinkMinimum(ctx, service, desired, imagePath)
+ var rejected *ErrShrinkRejected
+ if errors.As(err, &rejected) {
+ rejection, cleanupErr := m.recordShrinkRejection(dir, current, desired, rejected)
+ if cleanupErr != nil {
+ return nil, cleanupErr
+ }
+ return rejection, nil
+ }
+ if err != nil {
+ return nil, err
}
}
if direction == "grow" {
tx.Phase = "file_extended"
if err := writeJSONAtomic(transactionPath, tx); err != nil {
- return err
+ return nil, err
}
if err := os.Truncate(imagePath, desired.SizeBytes); err != nil {
- return fmt.Errorf("extend backing image: %w", err)
+ return nil, fmt.Errorf("extend backing image: %w", err)
}
- if _, err := m.runner.Run(ctx, "resize2fs", imagePath); err != nil {
- return err
+ if _, err := m.runner.RunDestructive(ctx, "resize2fs", imagePath); err != nil {
+ return nil, err
}
} else {
tx.Phase = "filesystem_shrinking"
if err := writeJSONAtomic(transactionPath, tx); err != nil {
- return err
+ return nil, err
}
- if _, err := m.runner.Run(ctx, "resize2fs", imagePath, strconv.FormatInt(desired.SizeBytes/1024, 10)+"K"); err != nil {
- return err
+ if _, err := m.runner.RunDestructive(ctx, "resize2fs", imagePath, strconv.FormatInt(desired.SizeBytes/1024, 10)+"K"); err != nil {
+ return nil, err
}
tx.Phase = "filesystem_shrunk"
if err := writeJSONAtomic(transactionPath, tx); err != nil {
- return err
+ return nil, err
}
if err := os.Truncate(imagePath, desired.SizeBytes); err != nil {
- return fmt.Errorf("truncate backing image after filesystem shrink: %w", err)
+ return nil, fmt.Errorf("truncate backing image after filesystem shrink: %w", err)
}
}
- if _, err := m.runner.Run(ctx, "e2fsck", "-f", "-y", imagePath); err != nil {
- return err
+ if _, err := m.runner.RunDestructive(ctx, "e2fsck", "-f", "-y", imagePath); err != nil {
+ return nil, err
}
current.AppliedSizeBytes = desired.SizeBytes
current.ResizeGeneration = desired.ResizeGeneration
+ // A size actually applied supersedes any earlier refusal.
+ current.clearRejection()
current.UpdatedAt = time.Now().UTC()
if err := writeJSONAtomic(filepath.Join(dir, manifestFilename), current); err != nil {
- return err
+ return nil, err
}
if err := os.Remove(transactionPath); err != nil && !os.IsNotExist(err) {
- return fmt.Errorf("remove resize transaction: %w", err)
+ return nil, fmt.Errorf("remove resize transaction: %w", err)
}
if err := syncDir(dir); err != nil {
- return fmt.Errorf("sync volume directory: %w", err)
+ return nil, fmt.Errorf("sync volume directory: %w", err)
}
- return nil
+ return nil, nil
+}
+
+// recordShrinkRejection cleans up the checking transaction and records the
+// refusal, in that order.
+//
+// The order is fixed and crash-consistent. Writing the rejection first risks
+// "rejection recorded plus stale checking transaction", which is exactly the
+// state that quarantines the corrected retry: prepareOne compares the stale
+// transaction's generation against the new one and refuses to proceed. Crashing
+// after the removal instead loses only the rejection record — the request is
+// re-measured, refused again, and recorded on the next pass, which is
+// idempotent and self-healing.
+//
+// Removing the transaction is safe here not because nothing has touched the
+// image (e2fsck ran, and may have replayed a journal) but because the checking
+// phase completes without changing the filesystem's *geometry*. No
+// partially-applied resize exists for the transaction to describe. Every later
+// phase has moved geometry, and its transaction must survive for recovery.
+func (m *Manager) recordShrinkRejection(dir string, current *manifest, desired config.VolumeConfig, rejected *ErrShrinkRejected) (*Rejection, error) {
+ if err := os.Remove(filepath.Join(dir, transactionFilename)); err != nil && !os.IsNotExist(err) {
+ return nil, fmt.Errorf("remove checking transaction after rejection: %w", err)
+ }
+ if err := syncDir(dir); err != nil {
+ return nil, err
+ }
+ current.RejectedGeneration = desired.ResizeGeneration
+ current.RejectedSizeBytes = desired.SizeBytes
+ current.RejectedMinimumBytes = rejected.Minimum
+ current.RejectedAt = time.Now().UTC()
+ current.UpdatedAt = current.RejectedAt
+ if err := writeJSONAtomic(filepath.Join(dir, manifestFilename), current); err != nil {
+ return nil, err
+ }
+ if err := syncDir(dir); err != nil {
+ return nil, err
+ }
+ rejection := current.rejectionFor(desired.ResizeGeneration)
+ return &rejection, nil
}
func (m *Manager) checkCapacity(pool *config.LocalStorageConfig, desired map[string]int64) error {
@@ -486,8 +890,14 @@ func (m *Manager) checkCapacity(pool *config.LocalStorageConfig, desired map[str
retained[id] = size
}
var reserved int64
- for _, size := range retained {
- if size > 0 && reserved > (1<<63-1)-size {
+ for id, size := range retained {
+ // Defence in depth: readRetained rejects a non-positive retained size,
+ // but the desired sizes merged in above come from a node config, and a
+ // reservation total that can be driven downwards is a capacity bypass.
+ if size <= 0 {
+ return fmt.Errorf("volume %s has non-positive size %d", id, size)
+ }
+ if reserved > (1<<63-1)-size {
return fmt.Errorf("local volume reservations overflow")
}
reserved += size
@@ -497,9 +907,11 @@ func (m *Manager) checkCapacity(pool *config.LocalStorageConfig, desired map[str
return fmt.Errorf("read local storage free space: %w", err)
}
available := int64(stat.Bavail) * int64(stat.Bsize)
- if m.observer != nil {
- m.observer.ObserveVolumePool(string(config.VolumeTypeLocal), reserved, pool.CapacityBytes, available)
- }
+ // Pool observation deliberately does not happen here. checkCapacity runs
+ // only when a service declares local volumes, so reporting from it made
+ // the gauges vanish on a node holding retained-but-unplaced volumes —
+ // exactly the state an operator needs them for. ObservePool now publishes
+ // them once per tick from the agent loop, independent of desired state.
if reserved > pool.CapacityBytes {
return fmt.Errorf("local volume capacity exceeded: reserved %d bytes, configured %d bytes", reserved, pool.CapacityBytes)
}
@@ -519,6 +931,214 @@ func (m *Manager) checkCapacity(pool *config.LocalStorageConfig, desired map[str
return nil
}
+// rebuildRejections replaces the whole refusal snapshot from the durable
+// manifests of the currently desired services. It is the complete
+// reconciliation; refreshRejections keeps one service fresh within a tick.
+func (m *Manager) rebuildRejections(services []config.ServiceConfig) {
+ if m == nil || m.storage.Local == nil {
+ return
+ }
+ rebuilt := make(map[string]Rejection)
+ for _, svc := range services {
+ for _, declared := range svc.Volumes {
+ if declared.Type != config.VolumeTypeLocal {
+ continue
+ }
+ // Evaluated against the *raw* desired config, before the clamp
+ // below rewrites it. After clamping, the size is the applied one
+ // and the request that was refused is no longer visible.
+ if rejection, refusing, _ := m.storedRejection(svc.Name, declared); refusing {
+ rebuilt[svc.Name+"/"+declared.Name] = rejection
+ }
+ }
+ }
+ m.rejectionMu.Lock()
+ defer m.rejectionMu.Unlock()
+ m.rejections = rebuilt
+}
+
+// storedRejection reads one volume's durable refusal and reports whether it
+// still describes the request being made. hasRecord distinguishes "no refusal
+// recorded at all" from "recorded, but no longer being requested", which the
+// callers need in order to prune correctly.
+func (m *Manager) storedRejection(service string, declared config.VolumeConfig) (rejection Rejection, refusing, hasRecord bool) {
+ dir, err := volumeDir(m.storage.Local.Path, service, declared.Name)
+ if err != nil {
+ return Rejection{}, false, false
+ }
+ var current manifest
+ if err := readJSON(filepath.Join(dir, manifestFilename), ¤t); err != nil || current.RejectedGeneration == 0 {
+ return Rejection{}, false, false
+ }
+ return current.rejectionFor(current.RejectedGeneration), current.refusesRequest(declared), true
+}
+
+// refreshRejections updates the refusal snapshot for one service from the
+// durable manifests.
+//
+// It reads the manifests rather than only the outcomes of this pass because
+// the clamp erases the evidence from the desired configuration: once the
+// effective size and generation are substituted, neither the preflight nor
+// prepareOne has anything left to refuse, and a snapshot built from outcomes
+// alone would clear itself on the very tick that proves the rejection is
+// working. The manifest is where the rejection actually lives, and a resize
+// that succeeds clears it there.
+func (m *Manager) refreshRejections(svc config.ServiceConfig) {
+ if m == nil || m.storage.Local == nil {
+ return
+ }
+ type outcome struct {
+ rejection Rejection
+ refusing bool
+ hasRecord bool
+ }
+ current := make(map[string]outcome, len(svc.Volumes))
+ for _, declared := range svc.Volumes {
+ if declared.Type != config.VolumeTypeLocal {
+ continue
+ }
+ rejection, refusing, hasRecord := m.storedRejection(svc.Name, declared)
+ current[svc.Name+"/"+declared.Name] = outcome{rejection, refusing, hasRecord}
+ }
+ m.rejectionMu.Lock()
+ defer m.rejectionMu.Unlock()
+ for logicalID, got := range current {
+ switch {
+ case !got.hasRecord:
+ // The refusal is gone from the manifest — a resize applied — so
+ // it stops being reported immediately rather than a tick later.
+ delete(m.rejections, logicalID)
+ case got.refusing:
+ m.rejections[logicalID] = got.rejection
+ }
+ // Otherwise leave the entry alone. By this point the config has
+ // already been normalized, so the refused size is no longer visible in
+ // it and this function cannot tell a withdrawn request from a standing
+ // one. rebuildRejections makes that call once per tick against the raw
+ // config; this pass only ever adds a refusal it has just discovered.
+ }
+}
+
+// Rejections returns the current refusal snapshot, keyed by logical ID.
+func (m *Manager) Rejections() map[string]Rejection {
+ if m == nil {
+ return nil
+ }
+ m.rejectionMu.RLock()
+ defer m.rejectionMu.RUnlock()
+ out := make(map[string]Rejection, len(m.rejections))
+ for id, rejection := range m.rejections {
+ out[id] = rejection
+ }
+ return out
+}
+
+// SeedRejectionsForTest installs a refusal snapshot directly. Production only
+// ever populates it from the durable manifests, through refreshRejections.
+func (m *Manager) SeedRejectionsForTest(rejections map[string]Rejection) {
+ if m == nil {
+ return
+ }
+ m.rejectionMu.Lock()
+ defer m.rejectionMu.Unlock()
+ m.rejections = make(map[string]Rejection, len(rejections))
+ for id, rejection := range rejections {
+ m.rejections[id] = rejection
+ }
+}
+
+// NormalizeVolumes rewrites a desired node configuration so every volume whose
+// exact request has already been refused renders its effective size instead.
+//
+// This closes the window before the control plane's own clamp catches up:
+// acknowledging a rejection and re-rendering takes at least one control-plane
+// cycle, and during that window the node config still carries the refused size.
+// Running the clamp here means needsUpdate, Prepare, and writeVMConfig all see
+// one configuration, and the instance stores that same configuration — so it
+// compares equal on the very next tick rather than one convergence cycle later.
+//
+// It reads the manifests rather than the in-memory snapshot so it is correct on
+// the first tick after an agent restart, when nothing has been measured yet.
+//
+// The match is on both generation and requested size (see manifest.matchesRejection),
+// which is what lets a hand-authored direct-Git config correct a refused shrink
+// by editing size_bytes alone.
+func (m *Manager) NormalizeVolumes(services []config.ServiceConfig) {
+ if m == nil || m.storage.Local == nil {
+ return
+ }
+ // Reconcile the refusal snapshot against the durable manifests for the
+ // whole desired set, not just the volumes some later Prepare happens to
+ // touch. Two things depend on it being complete:
+ //
+ // - after an agent restart the snapshot is empty, and if normalization
+ // clamps the config so that no action is planned, nothing else would
+ // ever repopulate it — the node would report every size applied while
+ // running an effective one;
+ // - a volume that is no longer declared has to drop out, or its stale
+ // entry keeps VolumeSizesApplied false forever.
+ m.rebuildRejections(services)
+ for si := range services {
+ for vi := range services[si].Volumes {
+ volume := &services[si].Volumes[vi]
+ if volume.Type != config.VolumeTypeLocal {
+ continue
+ }
+ dir, err := volumeDir(m.storage.Local.Path, services[si].Name, volume.Name)
+ if err != nil {
+ continue
+ }
+ var current manifest
+ if err := readJSON(filepath.Join(dir, manifestFilename), ¤t); err != nil {
+ continue
+ }
+ if !current.matchesRejection(*volume) {
+ continue
+ }
+ // Substitute the whole effective configuration — size *and*
+ // generation. Clamping only the size leaves the generation
+ // differing forever, and needsUpdate compares whole volume
+ // configs: the service would be re-planned on every tick, which is
+ // exactly the loop this is here to end. The refused request is not
+ // lost: it is reported from the rejection snapshot, which is where
+ // the acknowledgement reads it.
+ volume.SizeBytes = current.AppliedSizeBytes
+ volume.ResizeGeneration = current.ResizeGeneration
+ }
+ }
+}
+
+// ObservePool publishes the local pool gauges from retained state alone. It is
+// called once per agent tick, independent of any desired configuration, so a
+// node with retained but unplaced volumes — or with no desired local volumes at
+// all — keeps reporting reserved, capacity, and available bytes.
+//
+// It never fails a tick: a pool that is not configured or not readable is
+// simply not reported, because a metrics side effect must not be able to block
+// reconciliation.
+func (m *Manager) ObservePool() {
+ if m == nil || m.observer == nil || m.storage.Local == nil {
+ return
+ }
+ pool := m.storage.Local
+ retained, err := readRetained(pool.Path)
+ if err != nil {
+ return
+ }
+ var reserved int64
+ for _, size := range retained {
+ if size > 0 && reserved > (1<<63-1)-size {
+ return
+ }
+ reserved += size
+ }
+ var stat syscall.Statfs_t
+ if err := syscall.Statfs(pool.Path, &stat); err != nil {
+ return
+ }
+ m.observer.ObserveVolumePool(string(config.VolumeTypeLocal), reserved, pool.CapacityBytes, int64(stat.Bavail)*int64(stat.Bsize))
+}
+
func readRetained(root string) (map[string]int64, error) {
retained := make(map[string]int64)
err := filepath.WalkDir(root, func(path string, entry os.DirEntry, walkErr error) error {
@@ -535,7 +1155,28 @@ func readRetained(root string) (map[string]int64, error) {
if err := readJSON(path, &m); err != nil {
return fmt.Errorf("read retained manifest %s: %w", path, err)
}
- retained[m.LogicalID] = m.AppliedSizeBytes
+ // A retained size feeds pool arithmetic directly, and a non-positive
+ // one *subtracts* from the reserved total — so a single corrupt or
+ // hand-edited manifest can admit a volume the pool cannot hold. There
+ // is no safe number to assume for it, so it fails closed.
+ if m.AppliedSizeBytes <= 0 {
+ return fmt.Errorf("retained manifest %s has non-positive applied size %d; quarantined", path, m.AppliedSizeBytes)
+ }
+ // The map key must come from the volume's location, not from what its
+ // manifest claims to be. Keying by the declared logical ID lets two
+ // manifests collapse onto one entry, and the second volume's bytes
+ // disappear from the reserved total — the same capacity bypass a
+ // negative size produces, by a different route. A disagreement between
+ // the two is itself the corruption, so it fails closed.
+ located, err := logicalIDFromManifestPath(root, path)
+ if err != nil {
+ return err
+ }
+ if m.LogicalID != located {
+ return fmt.Errorf("retained manifest %s declares logical id %q but is stored at %q; quarantined",
+ path, m.LogicalID, located)
+ }
+ retained[located] = m.AppliedSizeBytes
return nil
})
if os.IsNotExist(err) {
@@ -544,6 +1185,74 @@ func readRetained(root string) (map[string]int64, error) {
return retained, err
}
+// ValidateNodeVolumes checks every volume declaration in a node config against
+// the invariants the agent enforces before it will run them.
+//
+// It is exported for `configcheck --node-config`, which validates hand-authored
+// direct-Git configs. It deliberately reuses the agent's own rules rather than
+// restating them: a second copy would drift, and the failure mode of drift here
+// is a config that validates in CI and then cannot start on the node.
+//
+// It checks declarations only. Anything requiring the host — a retained
+// manifest, pool capacity, free space, the node's own identity — is not
+// knowable from a config file and is left to the agent's Preflight.
+func ValidateNodeVolumes(nc config.NodeConfig) error {
+ var problems []string
+ for _, svc := range nc.Services {
+ if len(svc.Volumes) == 0 {
+ continue
+ }
+ // The agent derives a filesystem path from the service name as well as
+ // the volume name, so a name that is not a safe path component fails at
+ // preflight. It shares volumeDir's predicate rather than restating it,
+ // so the two cannot drift.
+ if err := validatePathComponent("service", svc.Name); err != nil {
+ problems = append(problems, err.Error())
+ continue
+ }
+ if err := validateServiceVolumes(svc.Volumes); err != nil {
+ problems = append(problems, fmt.Sprintf("service %s: %v", svc.Name, err))
+ continue
+ }
+ for _, declared := range svc.Volumes {
+ logicalID := svc.Name + "/" + declared.Name
+ switch declared.Type {
+ case config.VolumeTypeLocal:
+ // A local volume is durably bound to one physical node, and
+ // the agent refuses any volume whose bound_node does not match
+ // its own stable node_id. A config that omits it can never
+ // start.
+ if declared.BoundNode == "" {
+ problems = append(problems, fmt.Sprintf(
+ "volume %s: local volumes must declare bound_node matching the agent's node_id", logicalID))
+ }
+ case config.VolumeTypeShared:
+ problems = append(problems, fmt.Sprintf("volume %s: %v", logicalID, ErrSharedUnsupported))
+ default:
+ problems = append(problems, fmt.Sprintf("volume %s: unsupported type %q", logicalID, declared.Type))
+ }
+ }
+ }
+ if len(problems) > 0 {
+ return fmt.Errorf("%s", strings.Join(problems, "\n"))
+ }
+ return nil
+}
+
+// logicalIDFromManifestPath recovers "service/volume" from a manifest's
+// location under the pool root, which is the volume's real identity.
+func logicalIDFromManifestPath(root, path string) (string, error) {
+ rel, err := filepath.Rel(root, path)
+ if err != nil {
+ return "", fmt.Errorf("locate retained manifest %s: %w", path, err)
+ }
+ parts := strings.Split(filepath.ToSlash(rel), "/")
+ if len(parts) != 3 || parts[0] == "" || parts[1] == "" {
+ return "", fmt.Errorf("retained manifest %s is not at