server-deploy/internal/state/operation_test.go

package state

import (
	"errors"
	"fmt"
	"os"
	"path/filepath"
	"strings"
	"sync"
	"sync/atomic"
	"testing"
)

func TestIdempotencyAndTransitions(t *testing.T) {
	dir := t.TempDir()
	s, err := Acquire(dir, "host-one")
	if err != nil {
		t.Fatal(err)
	}
	defer s.Close()
	op, created, err := s.Begin("op-one", testHash)
	if err != nil || !created || op.Status != Queued || op.Revision != 1 {
		t.Fatalf("begin: %+v %v", op, err)
	}
	duplicate, created, err := s.Begin("op-one", testHash)
	if err != nil || created || duplicate != op {
		t.Fatal("duplicate mutated operation")
	}
	if _, _, err := s.Begin("op-one", "sha256:"+strings.Repeat("b", 64)); !errors.Is(err, ErrConflict) {
		t.Fatal("id rebound to another plan")
	}
	if _, _, err := s.Begin("op-two", testHash); !errors.Is(err, ErrUnresolved) {
		t.Fatal("unresolved operation bypass")
	}
	if _, err := s.Advance(op.ID, op.Revision, Succeeded); !errors.Is(err, ErrTransition) {
		t.Fatal("queued operation succeeded without running")
	}
	op, err = s.Advance(op.ID, op.Revision, Running)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := s.Advance(op.ID, 1, Succeeded); !errors.Is(err, ErrConflict) {
		t.Fatal("stale revision accepted")
	}
	if _, err := s.Advance(op.ID, op.Revision, Running); !errors.Is(err, ErrTransition) {
		t.Fatal("operation claimed twice")
	}
	op, err = s.Advance(op.ID, op.Revision, NeedsAttention)
	if err != nil {
		t.Fatal(err)
	}
	if _, _, err := s.Begin("op-two", testHash); !errors.Is(err, ErrUnresolved) {
		t.Fatal("ignored manual recovery")
	}
	op, err = s.Advance(op.ID, op.Revision, FailedRecovered)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := s.Advance(op.ID, op.Revision, Running); !errors.Is(err, ErrTransition) {
		t.Fatal("terminal operation restarted")
	}
	s.Close()
	s, err = Acquire(dir, "host-one")
	if err != nil {
		t.Fatal(err)
	}
	defer s.Close()
	got, err := s.Get("op-one")
	if err != nil || got != op {
		t.Fatalf("persistence mismatch: %+v %v", got, err)
	}
	if _, created, err := s.Begin("op-two", testHash); err != nil || !created {
		t.Fatal("completed operation blocks new work")
	}
}

func TestCorruptionAndHostMismatchFailClosed(t *testing.T) {
	dir := t.TempDir()
	s, err := Acquire(dir, "host-one")
	if err != nil {
		t.Fatal(err)
	}
	if _, _, err := s.Begin("op-one", testHash); err != nil {
		t.Fatal(err)
	}
	s.Close()
	if other, err := Acquire(dir, "host-two"); err == nil {
		other.Close()
		t.Fatal("wrong host accepted")
	}
	for _, data := range []string{`{`, `{}`, `null`, strings.Repeat("x", maxSnapshotBytes+1)} {
		if err := os.WriteFile(filepath.Join(dir, "state.json"), []byte(data), 0600); err != nil {
			t.Fatal(err)
		}
		if other, err := Acquire(dir, "host-one"); err == nil {
			other.Close()
			t.Fatal("corrupt state silently reset")
		}
	}
}

func TestInvalidOperationsDoNotCreateSnapshot(t *testing.T) {
	dir := t.TempDir()
	s, err := Acquire(dir, "host-one")
	if err != nil {
		t.Fatal(err)
	}
	defer s.Close()
	for _, id := range []string{"", "../outside", "bad/id"} {
		if _, _, err := s.Begin(id, testHash); err == nil {
			t.Fatal("invalid operation id accepted")
		}
	}
	if _, _, err := s.Begin("op-one", "latest"); err == nil {
		t.Fatal("mutable plan binding accepted")
	}
	if _, err := s.Get("missing"); !errors.Is(err, ErrNotFound) {
		t.Fatal("missing operation not reported")
	}
	if _, err := os.Stat(filepath.Join(dir, "state.json")); !os.IsNotExist(err) {
		t.Fatal("invalid request persisted state")
	}
}

func TestPersistenceFailurePoisonsSession(t *testing.T) {
	dir := t.TempDir()
	s, err := Acquire(dir, "host-one")
	if err != nil {
		t.Fatal(err)
	}
	defer s.Close()
	// A directory at the destination makes atomic replacement fail on both OSes.
	if err := os.Mkdir(filepath.Join(dir, "state.json"), 0700); err != nil {
		t.Fatal(err)
	}
	if _, _, err := s.Begin("op-one", testHash); err == nil {
		t.Fatal("reported failed persistence as success")
	}
	if _, _, err := s.Begin("op-two", testHash); !errors.Is(err, ErrPoisoned) {
		t.Fatalf("continued after ambiguous write: %v", err)
	}
}

func TestConcurrentClaimsHaveOneWinner(t *testing.T) {
	s, err := Acquire(t.TempDir(), "host-one")
	if err != nil {
		t.Fatal(err)
	}
	defer s.Close()
	var createdCount, claimedCount atomic.Int32
	var wg sync.WaitGroup
	for n := 0; n < 16; n++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			_, created, err := s.Begin("op-one", testHash)
			if err != nil {
				t.Error(err)
			}
			if created {
				createdCount.Add(1)
			}
		}()
	}
	wg.Wait()
	for n := 0; n < 16; n++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			_, err := s.Advance("op-one", 1, Running)
			if err == nil {
				claimedCount.Add(1)
			} else if !errors.Is(err, ErrConflict) {
				t.Error(err)
			}
		}()
	}
	wg.Wait()
	if createdCount.Load() != 1 || claimedCount.Load() != 1 {
		t.Fatalf("duplicate winners: %d %d", createdCount.Load(), claimedCount.Load())
	}
}

func TestTamperedSnapshotRejected(t *testing.T) {
	dir := t.TempDir()
	s, err := Acquire(dir, "host-one")
	if err != nil {
		t.Fatal(err)
	}
	if _, _, err := s.Begin("op-one", testHash); err != nil {
		t.Fatal(err)
	}
	s.Close()
	path := filepath.Join(dir, "state.json")
	raw, err := os.ReadFile(path)
	if err != nil {
		t.Fatal(err)
	}
	for _, changed := range []string{
		strings.Replace(string(raw), `"queued"`, `"succeeded"`, 1),
		strings.Replace(string(raw), `"version":1`, `"version":1,"version":1`, 1),
		string(raw) + ` {}`,
	} {
		if err := os.WriteFile(path, []byte(changed), 0600); err != nil {
			t.Fatal(err)
		}
		if s, err := Acquire(dir, "host-one"); err == nil {
			s.Close()
			t.Fatal("tampered snapshot accepted")
		}
	}
}

func TestAllTerminalAndReconciliationPaths(t *testing.T) {
	for _, path := range [][]Status{{Cancelled}, {Running, Succeeded}, {Running, FailedRecovered}, {Running, Unknown, FailedRecovered}, {Running, NeedsAttention, Succeeded}} {
		s, err := Acquire(t.TempDir(), "host-one")
		if err != nil {
			t.Fatal(err)
		}
		op, _, err := s.Begin("op-one", testHash)
		if err != nil {
			t.Fatal(err)
		}
		for _, next := range path {
			op, err = s.Advance(op.ID, op.Revision, next)
			if err != nil {
				t.Fatal(err)
			}
		}
		if _, _, err := s.Begin("op-two", testHash); err != nil {
			t.Fatal(err)
		}
		s.Close()
	}
}

func TestMissingSnapshotDoesNotResetInitializedStore(t *testing.T) {
	dir := t.TempDir()
	s, err := Acquire(dir, "host-one")
	if err != nil {
		t.Fatal(err)
	}
	if _, _, err := s.Begin("op-one", testHash); err != nil {
		t.Fatal(err)
	}
	s.Close()
	if err := os.Remove(filepath.Join(dir, "state.json")); err != nil {
		t.Fatal(err)
	}
	if s, err := Acquire(dir, "host-one"); err == nil {
		s.Close()
		t.Fatal("silently reset initialized store")
	}
}

func TestAdmissionReservesSpaceForCompletion(t *testing.T) {
	dir := t.TempDir()
	data := snapshot{Version: 1, HostID: "host-one", Operations: make(map[string]Operation)}
	for n := 0; n < 113357; n++ {
		id := fmt.Sprintf("op%06d", n)
		if n < 2 {
			id += strings.Repeat("a", 26)
		}
		data.Operations[id] = Operation{ID: id, PlanHash: testHash, Status: Cancelled, Revision: 2}
	}
	raw, err := encodeSnapshot(data)
	if err != nil {
		t.Fatal(err)
	}
	if err := os.WriteFile(filepath.Join(dir, "state.json"), raw, 0600); err != nil {
		t.Fatal(err)
	}
	// Confirm this fixture reaches the actual bug boundary, not an arbitrary limit.
	data.Operations["op-new"] = Operation{ID: "op-new", PlanHash: testHash, Status: Queued, Revision: 1}
	queued, err := encodeSnapshot(data)
	if err != nil || len(queued) != maxSnapshotBytes-2 {
		t.Fatalf("fixture outside boundary: %d %v", len(queued), err)
	}
	s, err := Acquire(dir, "host-one")
	if err != nil {
		t.Fatal(err)
	}
	defer s.Close()
	if _, _, err := s.Begin("op-new", testHash); err == nil {
		t.Fatal("admitted task without room for terminal state")
	}
	if _, err := s.Get("op000002"); err != nil {
		t.Fatalf("capacity rejection poisoned read access: %v", err)
	}
	if _, err := s.Get("op-new"); !errors.Is(err, ErrNotFound) {
		t.Fatal("rejected admission persisted")
	}
}