EvoBGP/internal/store/deployment_test.go

package store

import (
	"context"
	"errors"
	"testing"
	"time"
)

func TestDeploymentLeaseGenerationAndReplay(t *testing.T) {
	m := NewMemory()
	m.SeedDemo()
	tenant, _, _, revision, speaker := m.DemoIDs()
	ctx := context.Background()
	enqueue := func() JobRecord {
		t.Helper()
		j, _, err := m.EnqueueJob(ctx, JobEnqueue{TenantID: tenant, Kind: "deploy_apply"})
		if err != nil {
			t.Fatal(err)
		}
		return j
	}
	claim := func(owner string) JobRecord {
		t.Helper()
		rows, err := m.ClaimJobs(ctx, owner, 1, time.Minute)
		if err != nil || len(rows) != 1 {
			t.Fatalf("claim: %v %v", rows, err)
		}
		return rows[0]
	}
	fence := func(j JobRecord) JobFence {
		return JobFence{JobID: j.ID, TenantID: j.TenantID, OwnerID: j.OwnerID, Token: j.FencingToken}
	}
	first := enqueue()
	first = claim("one")
	d, err := m.BeginDeploymentSpeaker(ctx, fence(first), speaker, revision)
	if err != nil {
		t.Fatal(err)
	}
	if err = m.RecordDeploymentSpeaker(ctx, fence(first), d, "applied", "verified", ""); !errors.Is(err, ErrInvalidInput) {
		t.Fatalf("unproved verified: %v", err)
	}
	if err = m.RecordDeploymentSpeaker(ctx, fence(first), d, "applied", "agent_reported", ""); err != nil {
		t.Fatal(err)
	}
	replayed, err := m.BeginDeploymentSpeaker(ctx, fence(first), speaker, revision)
	if err != nil || replayed.Generation != d.Generation || replayed.State != "applied" {
		t.Fatalf("replay: %+v %v", replayed, err)
	}
	if err = m.ReleaseJob(ctx, first.ID, first.OwnerID, first.FencingToken); err != nil {
		t.Fatal(err)
	}
	reclaimed := claim("two")
	if err = m.RecordDeploymentSpeaker(ctx, fence(first), d, "applied", "agent_reported", ""); !errors.Is(err, ErrJobLeaseLost) {
		t.Fatalf("old owner ACK: %v", err)
	}
	if _, err = m.BeginDeploymentSpeaker(ctx, fence(reclaimed), speaker, revision); err != nil {
		t.Fatal(err)
	}
	second := enqueue()
	second = claim("three")
	d2, err := m.BeginDeploymentSpeaker(ctx, fence(second), speaker, revision)
	if err != nil {
		t.Fatal(err)
	}
	if d2.Generation <= d.Generation {
		t.Fatal("generation failed to advance across independent job fencing tokens")
	}
	if err = m.RecordDeploymentSpeaker(ctx, fence(reclaimed), d, "applied", "agent_reported", ""); !errors.Is(err, ErrJobLeaseLost) {
		t.Fatalf("old generation ACK: %v", err)
	}
	if _, err = m.BeginDeploymentSpeaker(ctx, fence(reclaimed), speaker, revision); !errors.Is(err, ErrJobLeaseLost) {
		t.Fatalf("old generation republish: %v", err)
	}
	rows, err := m.ListDeploymentSpeakers(ctx, tenant, first.ID)
	if err != nil || len(rows) != 1 {
		t.Fatalf("durable progress: %v %v", rows, err)
	}
	if err = m.RecordDeploymentSpeaker(ctx, fence(second), d2, "unknown", "unconfirmed", "response lost"); err != nil {
		t.Fatal(err)
	}
}

func TestDeploymentObservationTokenRejectsLateHealth(t *testing.T) {
	m := NewMemory()
	m.SeedDemo()
	tenant, _, _, revision, speaker := m.DemoIDs()
	ctx := context.Background()
	_, _, err := m.EnqueueJob(ctx, JobEnqueue{TenantID: tenant, Kind: "deploy_apply"})
	if err != nil {
		t.Fatal(err)
	}
	jobs, err := m.ClaimJobs(ctx, "owner", 1, time.Minute)
	if err != nil || len(jobs) != 1 {
		t.Fatal(err)
	}
	j := jobs[0]
	f := JobFence{JobID: j.ID, TenantID: tenant, OwnerID: j.OwnerID, Token: j.FencingToken}
	d, err := m.BeginDeploymentSpeaker(ctx, f, speaker, revision)
	if err != nil {
		t.Fatal(err)
	}
	if err = m.RecordDeploymentSpeaker(ctx, f, d, "applied", "agent_reported", ""); err != nil {
		t.Fatal(err)
	}
	j.Status = "succeeded"
	if err = m.FinishJob(ctx, j); err != nil {
		t.Fatal(err)
	}
	old, err := m.ListDeploymentObservations(ctx, 1)
	if err != nil || len(old) != 1 {
		t.Fatal(err)
	}
	latest, err := m.ListDeploymentObservations(ctx, 1)
	if err != nil || len(latest) != 1 {
		t.Fatal(err)
	}
	if err = m.ObserveDeployment(ctx, latest[0], DeploymentObservation{Error: "timeout"}); err != nil {
		t.Fatal(err)
	}
	if err = m.ObserveDeployment(ctx, old[0], DeploymentObservation{RevisionID: "older"}); !errors.Is(err, ErrJobLeaseLost) {
		t.Fatalf("late report accepted: %v", err)
	}
	rows, err := m.ListDeploymentSpeakers(ctx, tenant, j.ID)
	if err != nil || rows[0].State != "applied" || rows[0].Confirmation != "agent_reported" || rows[0].HealthError != "timeout" {
		t.Fatalf("error downgraded confirmation: %+v %v", rows, err)
	}
}

func TestDeploymentReconciliationRetryBudget(t *testing.T) {
	m := NewMemory()
	m.SeedDemo()
	tenant, _, _, revision, speaker := m.DemoIDs()
	ctx := context.Background()
	var previous DeploymentSpeaker
	for round := 0; round <= 3; round++ {
		_, _, err := m.EnqueueJob(ctx, JobEnqueue{TenantID: tenant, Kind: "deploy_apply"})
		if err != nil {
			t.Fatal(err)
		}
		jobs, err := m.ClaimJobs(ctx, "worker", 1, time.Minute)
		if err != nil || len(jobs) != 1 {
			t.Fatal(err)
		}
		j := jobs[0]
		f := JobFence{JobID: j.ID, TenantID: tenant, OwnerID: j.OwnerID, Token: j.FencingToken, RetryRound: round}
		if round > 0 {
			f.ExpectedGeneration = previous.Generation
		}
		d, err := m.BeginDeploymentSpeaker(ctx, f, speaker, revision)
		if err != nil {
			t.Fatal(err)
		}
		if err = m.RecordDeploymentSpeaker(ctx, f, d, "failed", "unconfirmed", "unreachable"); err != nil {
			t.Fatal(err)
		}
		j.Status = "failed"
		if err = m.FinishJob(ctx, j); err != nil {
			t.Fatal(err)
		}
		pending, err := m.ListPendingDeployments(ctx, 100)
		if err != nil {
			t.Fatal(err)
		}
		if (round < 3 && len(pending) != 1) || (round == 3 && len(pending) != 0) {
			t.Fatalf("round %d pending=%d", round, len(pending))
		}
		previous = d
	}
}