gate failover on replica freshness

Author: seslly

README.md

@@ -99,12 +99,19 @@ metadata:
     simple-volume.shipstuff.io/failover-workload-kind: "Deployment"
     simple-volume.shipstuff.io/failover-workload-name: "my-writer"
     simple-volume.shipstuff.io/failover-grace-period: "1m"
+    simple-volume.shipstuff.io/failover-max-staleness: "2m"
 ```
 
 When no healthy pod using the claim is running on a Ready, schedulable node for
-the grace period, the controller selects a ready replica node, patches the
-Deployment's `kubernetes.io/hostname` node selector, and deletes stale pods
-using the claim.
+the grace period, the controller selects a ready replica node whose last
+successful sync is within `failover-max-staleness`, patches the Deployment's
+`kubernetes.io/hostname` node selector, and deletes stale pods using the claim.
+
+The old active node is not promoted back automatically when it returns. If the
+old active rejoins as a replica target, the next full sync asks that agent to
+move its existing local volume into `.simple-volume-backups/` before restoring
+from the current leader. That keeps a rollback copy of the pre-restore local
+state while avoiding split-brain.
 
 ## Development
 

docs/e2e-deployment-tasks.md

@@ -82,6 +82,9 @@ kubectl exec deploy/simple-volume-demo -- sh -c 'echo ok >> /data/e2e.txt && cat
 - Add an E2E test script that writes data, syncs it to a second node, simulates
   active-node loss, promotes a fresh replica, reschedules the demo pod, and
   verifies data on the promoted node.
+- Extend the failover drill so the old active node rejoins, backs up its stale
+  local copy under `.simple-volume-backups/`, restores from the promoted leader,
+  and remains a replica unless an explicit move-back is requested.
 
 Recommended V0 replication policy shape:
 

docs/v0-plan.md

@@ -78,6 +78,14 @@
       - target replica within maxStaleness
       - target agent healthy
       - old writer marked demoted/stale in API state
+      - returning old active is treated as stale local state, not as the leader
+
+  - Returning-node restore:
+      - promoted node remains the source of truth
+      - old active rejoins as a replica target only
+      - before restore, the node agent moves its existing local volume into a timestamped `.simple-volume-backups/` path
+      - restore then overwrites the live replica path from the current leader
+      - automatic move-back is out of scope; planned move-back must be explicit
 
   - Auto-promotion is opt-in per volume and controlled by per-volume grace/deadline fields.
   - Workload movement uses normal Kubernetes scheduling:
@@ -128,4 +136,3 @@
   - CSI is a thin Kubernetes volume boundary, not the replication engine.
   - Node-agent DaemonSet placement defines the default storage-capable node set.
   - Production workload adoption requires a later explicit rollout plan, restore drill, and failure-mode test.
-

internal/agent/agent.go

@@ -8,6 +8,7 @@ import (
 	"os"
 	"path/filepath"
 	"strings"
+	"time"
 )
 
 const PoolMarkerFile = ".simple-volume-pool"
@@ -88,6 +89,35 @@ func EnsureVolumePath(v VolumePath, perm os.FileMode) (string, error) {
 	return path, nil
 }
 
+func BackupVolumePath(v VolumePath, now time.Time) (string, error) {
+	path, err := ResolveVolumePath(v)
+	if err != nil {
+		return "", err
+	}
+	if _, err := os.Stat(path); os.IsNotExist(err) {
+		return "", nil
+	} else if err != nil {
+		return "", err
+	}
+	empty, err := IsEmptyDir(path)
+	if err != nil {
+		return "", err
+	}
+	if empty {
+		return "", nil
+	}
+	pool := v.Pool.Clean()
+	backupRoot := filepath.Join(pool.Path, ".simple-volume-backups", safeSegment(v.Namespace), safeSegment(v.Name))
+	if err := os.MkdirAll(backupRoot, 0o755); err != nil {
+		return "", err
+	}
+	backupPath := filepath.Join(backupRoot, now.UTC().Format("20060102T150405.000000000Z"))
+	if err := os.Rename(path, backupPath); err != nil {
+		return "", err
+	}
+	return backupPath, nil
+}
+
 type SyncRequest struct {
 	Method     string
 	SourceURL  string

internal/agent/replication.go

@@ -41,11 +41,12 @@ type SourceRef struct {
 }
 
 type FullSyncRequest struct {
-	Namespace    string    `json:"namespace,omitempty"`
-	Volume       string    `json:"volume"`
-	Source       SourceRef `json:"source"`
-	IncludePaths []string  `json:"includePaths,omitempty"`
-	ExcludePaths []string  `json:"excludePaths,omitempty"`
+	Namespace      string    `json:"namespace,omitempty"`
+	Volume         string    `json:"volume"`
+	Source         SourceRef `json:"source"`
+	IncludePaths   []string  `json:"includePaths,omitempty"`
+	ExcludePaths   []string  `json:"excludePaths,omitempty"`
+	BackupExisting bool      `json:"backupExisting,omitempty"`
 }
 
 type PathFilter struct {
@@ -223,11 +224,17 @@ func ApplyFullSync(ctx context.Context, runner Runner, pool Pool, req FullSyncRe
 	if req.Source.WebDAVURL == "" {
 		return fmt.Errorf("source.webdavUrl is required")
 	}
-	targetRoot, err := EnsureVolumePath(VolumePath{
+	volumePath := VolumePath{
 		Pool:      pool,
 		Namespace: req.Namespace,
 		Name:      req.Volume,
-	}, 0o755)
+	}
+	if req.BackupExisting {
+		if _, err := BackupVolumePath(volumePath, time.Now().UTC()); err != nil {
+			return err
+		}
+	}
+	targetRoot, err := EnsureVolumePath(volumePath, 0o755)
 	if err != nil {
 		return err
 	}

internal/agent/replication_test.go

@@ -135,6 +135,41 @@ func TestApplyFullSyncRunsScopedRcloneSync(t *testing.T) {
 	}
 }
 
+func TestApplyFullSyncCanBackupExistingTarget(t *testing.T) {
+	dir := t.TempDir()
+	pool := Pool{Name: "default", Path: dir}
+	if err := EnsurePool(pool, false); err != nil {
+		t.Fatal(err)
+	}
+	target := filepath.Join(dir, "default", "demo", "save.dat")
+	if err := writeFile(target, "old state"); err != nil {
+		t.Fatal(err)
+	}
+	runner := &recordingRunner{}
+	err := ApplyFullSync(context.Background(), runner, pool, FullSyncRequest{
+		Namespace:      "default",
+		Volume:         "demo",
+		Source:         SourceRef{WebDAVURL: "http://source:8081"},
+		BackupExisting: true,
+	})
+	if err != nil {
+		t.Fatalf("ApplyFullSync returned error: %v", err)
+	}
+	backups, err := filepath.Glob(filepath.Join(dir, ".simple-volume-backups", "default", "demo", "*", "save.dat"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(backups) != 1 {
+		t.Fatalf("backups = %#v, want one backed up save.dat", backups)
+	}
+	if _, err := os.Stat(filepath.Join(dir, "default", "demo")); err != nil {
+		t.Fatalf("restored target directory missing: %v", err)
+	}
+	if len(runner.specs) != 1 {
+		t.Fatalf("ran %d commands, want 1", len(runner.specs))
+	}
+}
+
 func writeFile(path, content string) error {
 	if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil {
 		return err

internal/agent/watch_manager.go

@@ -33,23 +33,34 @@ type WatchStopRequest struct {
 }
 
 type WatchStatus struct {
-	Namespace            string      `json:"namespace,omitempty"`
-	Volume               string      `json:"volume"`
-	Source               SourceRef   `json:"source"`
-	Targets              []TargetRef `json:"targets"`
-	IncludePaths         []string    `json:"includePaths,omitempty"`
-	ExcludePaths         []string    `json:"excludePaths,omitempty"`
-	Running              bool        `json:"running"`
-	StartedAt            time.Time   `json:"startedAt"`
-	StoppedAt            *time.Time  `json:"stoppedAt,omitempty"`
-	LastBatchAt          *time.Time  `json:"lastBatchAt,omitempty"`
-	LastBatchEventCount  int         `json:"lastBatchEventCount,omitempty"`
-	LastBatchGeneration  string      `json:"lastBatchGeneration,omitempty"`
-	LastDeliveryError    string      `json:"lastDeliveryError,omitempty"`
-	LastWatchError       string      `json:"lastWatchError,omitempty"`
-	DeliveredBatchCount  int64       `json:"deliveredBatchCount,omitempty"`
-	DeliveredEventCount  int64       `json:"deliveredEventCount,omitempty"`
-	DeliveryFailureCount int64       `json:"deliveryFailureCount,omitempty"`
+	Namespace            string         `json:"namespace,omitempty"`
+	Volume               string         `json:"volume"`
+	Source               SourceRef      `json:"source"`
+	Targets              []TargetRef    `json:"targets"`
+	TargetStatuses       []TargetStatus `json:"targetStatuses,omitempty"`
+	IncludePaths         []string       `json:"includePaths,omitempty"`
+	ExcludePaths         []string       `json:"excludePaths,omitempty"`
+	Running              bool           `json:"running"`
+	StartedAt            time.Time      `json:"startedAt"`
+	StoppedAt            *time.Time     `json:"stoppedAt,omitempty"`
+	LastBatchAt          *time.Time     `json:"lastBatchAt,omitempty"`
+	LastBatchEventCount  int            `json:"lastBatchEventCount,omitempty"`
+	LastBatchGeneration  string         `json:"lastBatchGeneration,omitempty"`
+	LastDeliveryError    string         `json:"lastDeliveryError,omitempty"`
+	LastWatchError       string         `json:"lastWatchError,omitempty"`
+	DeliveredBatchCount  int64          `json:"deliveredBatchCount,omitempty"`
+	DeliveredEventCount  int64          `json:"deliveredEventCount,omitempty"`
+	DeliveryFailureCount int64          `json:"deliveryFailureCount,omitempty"`
+}
+
+type TargetStatus struct {
+	URL                    string     `json:"url"`
+	LastSuccessfulSync     *time.Time `json:"lastSuccessfulSync,omitempty"`
+	LastObservedGeneration string     `json:"lastObservedGeneration,omitempty"`
+	LastError              string     `json:"lastError,omitempty"`
+	DeliveredBatchCount    int64      `json:"deliveredBatchCount,omitempty"`
+	DeliveredEventCount    int64      `json:"deliveredEventCount,omitempty"`
+	DeliveryFailureCount   int64      `json:"deliveryFailureCount,omitempty"`
 }
 
 type BatchSender interface {
@@ -139,14 +150,15 @@ func (m *WatchManager) Start(ctx context.Context, req WatchStartRequest) (WatchS
 	key := watchKey(req.Namespace, req.Volume)
 	watchCtx, cancel := context.WithCancel(context.Background())
 	status := WatchStatus{
-		Namespace:    req.Namespace,
-		Volume:       req.Volume,
-		Source:       req.Source,
-		Targets:      append([]TargetRef(nil), req.Targets...),
-		IncludePaths: append([]string(nil), req.IncludePaths...),
-		ExcludePaths: append([]string(nil), req.ExcludePaths...),
-		Running:      true,
-		StartedAt:    time.Now().UTC(),
+		Namespace:      req.Namespace,
+		Volume:         req.Volume,
+		Source:         req.Source,
+		Targets:        append([]TargetRef(nil), req.Targets...),
+		TargetStatuses: initialTargetStatuses(req.Targets),
+		IncludePaths:   append([]string(nil), req.IncludePaths...),
+		ExcludePaths:   append([]string(nil), req.ExcludePaths...),
+		Running:        true,
+		StartedAt:      time.Now().UTC(),
 	}
 
 	m.mu.Lock()
@@ -283,17 +295,17 @@ func (m *WatchManager) run(ctx context.Context, key string, watch *managedWatch,
 		batch.Source = source
 		for _, target := range targets {
 			if err := m.sender.SendBatch(ctx, target, batch); err != nil {
-				m.recordDeliveryError(key, watch, err)
+				m.recordDeliveryError(key, watch, target, err)
 				continue
 			}
-			m.recordDeliverySuccess(key, watch, batch)
+			m.recordDeliverySuccess(key, watch, target, batch)
 		}
 		return nil
 	})
 	m.recordStopped(key, watch, err)
 }
 
-func (m *WatchManager) recordDeliverySuccess(key string, watch *managedWatch, batch EventBatch) {
+func (m *WatchManager) recordDeliverySuccess(key string, watch *managedWatch, target TargetRef, batch EventBatch) {
 	now := time.Now().UTC()
 	m.mu.Lock()
 	defer m.mu.Unlock()
@@ -306,16 +318,25 @@ func (m *WatchManager) recordDeliverySuccess(key string, watch *managedWatch, ba
 	watch.status.DeliveredBatchCount++
 	watch.status.DeliveredEventCount += int64(len(batch.Events))
 	watch.status.LastDeliveryError = ""
+	targetStatus := ensureTargetStatus(&watch.status, target.URL)
+	targetStatus.LastSuccessfulSync = &now
+	targetStatus.LastObservedGeneration = batch.Generation
+	targetStatus.DeliveredBatchCount++
+	targetStatus.DeliveredEventCount += int64(len(batch.Events))
+	targetStatus.LastError = ""
 }
 
-func (m *WatchManager) recordDeliveryError(key string, watch *managedWatch, err error) {
+func (m *WatchManager) recordDeliveryError(key string, watch *managedWatch, target TargetRef, err error) {
 	m.mu.Lock()
 	defer m.mu.Unlock()
 	if m.watches[key] != watch {
 		return
 	}
 	watch.status.DeliveryFailureCount++
 	watch.status.LastDeliveryError = err.Error()
+	targetStatus := ensureTargetStatus(&watch.status, target.URL)
+	targetStatus.DeliveryFailureCount++
+	targetStatus.LastError = err.Error()
 }
 
 func (m *WatchManager) recordStopped(key string, watch *managedWatch, err error) {
@@ -350,7 +371,26 @@ func parseOptionalDuration(value string, fallback time.Duration) (time.Duration,
 
 func cloneWatchStatus(status WatchStatus) WatchStatus {
 	status.Targets = append([]TargetRef(nil), status.Targets...)
+	status.TargetStatuses = append([]TargetStatus(nil), status.TargetStatuses...)
 	status.IncludePaths = append([]string(nil), status.IncludePaths...)
 	status.ExcludePaths = append([]string(nil), status.ExcludePaths...)
 	return status
 }
+
+func initialTargetStatuses(targets []TargetRef) []TargetStatus {
+	statuses := make([]TargetStatus, 0, len(targets))
+	for _, target := range targets {
+		statuses = append(statuses, TargetStatus{URL: target.URL})
+	}
+	return statuses
+}
+
+func ensureTargetStatus(status *WatchStatus, targetURL string) *TargetStatus {
+	for i := range status.TargetStatuses {
+		if status.TargetStatuses[i].URL == targetURL {
+			return &status.TargetStatuses[i]
+		}
+	}
+	status.TargetStatuses = append(status.TargetStatuses, TargetStatus{URL: targetURL})
+	return &status.TargetStatuses[len(status.TargetStatuses)-1]
+}