294 lines
7.8 KiB
Go
294 lines
7.8 KiB
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")
|
|
}
|
|
}
|