Janady13/missionmatch / internal/integration/ingest_during_promotion_test.go

Source identity

Field	Recorded value
Source type	GitHub source
Original record	Janady13/missionmatch / internal/integration/ingest_during_promotion_test.go
Date	Not specified in source
Repository	Janady13/missionmatch
Revision	720d47fae476b0d38477ae4401697d372b1ea2bc
Branch	feature/opportunity-to-outcome
Path or archive member	internal/integration/ingest_during_promotion_test.go
Original SHA-256	f88c4bee85ef307de2cfb5ebfcb4ec4896a075db062185975f5938453635502c
Git blob SHA	c7f14eaf3d472472432fe8127889877a7f14d112
Reading-copy SHA-256	f88c4bee85ef307de2cfb5ebfcb4ec4896a075db062185975f5938453635502c
Original source ↗
Recovered source transcript

Download complete source text ↓310 lines · 12.3 KB
//go:build integration

package integration

import (
	"context"
	"fmt"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	"github.com/google/uuid"
)

// GATE 2, THE LAST FUNCTIONAL ITEM: an ingest run during promotion corrupts nothing, and approved
// requirements are never rewritten by any ingest path.
//
// This is the hinge of the whole safety boundary. Promotion is the one transition that makes machine
// derived work visible to the frozen scoring engine, and ingest is a background job that rewrites the
// public cache on a timer. If the two can interleave badly, then a solicitation refresh landing at the
// wrong moment could silently alter a requirement a named human already approved, and the AEDT audit trail
// would be describing something that is no longer true.
//
// It is proven by RACING it rather than by reasoning about it, which is the house rule.

// seedPromotionFixture builds one organization with a tracked opportunity and a requirement draft ready to
// approve. Everything is written through the BYPASSRLS migrator, because the test is about concurrency and
// not about tenancy, which the RLS suite already covers.
func seedPromotionFixture(t *testing.T) (org, opp, draft uuid.UUID) {
	t.Helper()
	ctx := context.Background()
	org = seedOrg(t)
	opp, draft = uuid.New(), uuid.New()

	notice := "RACE-" + opp.String()[:8]
	if _, err := migrator.Exec(ctx, `
		INSERT INTO sam_opportunities (notice_id, title, department, naics_code, raw, source)
		VALUES ($1, 'Cyber Systems Operations Support', 'DEPT OF THE AIR FORCE', '541512', '{}'::jsonb, 'test')
		ON CONFLICT (notice_id) DO NOTHING`, notice); err != nil {
		t.Fatalf("seed public cache: %v", err)
	}
	if _, err := migrator.Exec(ctx, `
		INSERT INTO tracked_opportunities (organization_id, id, sam_notice_id, source_type, title, status)
		VALUES ($1, $2, $3, 'sam_gov', 'Cyber Systems Operations Support', 'qualifying')`,
		org, opp, notice); err != nil {
		t.Fatalf("seed tracked opportunity: %v", err)
	}
	if _, err := migrator.Exec(ctx, `
		INSERT INTO requirement_drafts (organization_id, id, opportunity_id, labor_category, quantity,
		                                clearance_level, review_status, derivation_method)
		VALUES ($1, $2, $3, 'Cyber Systems Operator', 2, 'ts_sci', 'in_review', 'model')`,
		org, draft, opp); err != nil {
		t.Fatalf("seed requirement draft: %v", err)
	}
	t.Cleanup(func() {
		_, _ = migrator.Exec(context.Background(), `DELETE FROM requirement_drafts WHERE organization_id=$1`, org)
		_, _ = migrator.Exec(context.Background(), `DELETE FROM tracked_opportunities WHERE organization_id=$1`, org)
		_, _ = migrator.Exec(context.Background(), `DELETE FROM sam_opportunities WHERE notice_id=$1`, notice)
	})
	return org, opp, draft
}

// simulateIngest does what sam-sync's refresh does to the public cache: upsert the notice with changed
// source fields. It deliberately hammers the SAME notice the promotion is reading through.
func simulateIngest(ctx context.Context, notice string, round int) error {
	_, err := migrator.Exec(ctx, `
		UPDATE sam_opportunities
		   SET title = $2, synced_at = now(), raw = jsonb_build_object('round', $3::int)
		 WHERE notice_id = $1`,
		notice, fmt.Sprintf("Cyber Systems Operations Support REV %d", round), round)
	return err
}

// THE RACE. Eight concurrent promotions of one draft while an ingest loop rewrites the underlying notice
// continuously. Exactly one requirement must exist afterwards, and the draft must be approved exactly once
// with a named reviewer.
func TestIngestDuringPromotionElectsOneWinnerAndCorruptsNothing(t *testing.T) {
	if migrator == nil {
		t.Skip("integration pools not configured")
	}
	ctx := context.Background()
	org, opp, draft := seedPromotionFixture(t)

	var notice string
	if err := migrator.QueryRow(ctx, `SELECT sam_notice_id FROM tracked_opportunities WHERE id=$1`, opp).Scan(&notice); err != nil {
		t.Fatalf("read notice: %v", err)
	}

	// The reviewer is seeded BEFORE the race, deliberately. Seeding inside a racer would have each goroutine
	// hold a transaction connection AND request a second one from the same pool, which deadlocks on pool
	// exhaustion the moment the racer count reaches the pool size. That is a test bug that looks exactly
	// like the production deadlock it is meant to detect, so it is worth avoiding by construction.
	reviewerID, _ := seedRoleUser(t, org, "approver", "manager")

	// The ingest loop runs for the whole duration of the promotion race.
	stop := make(chan struct{})
	var ingestRounds int64
	var ingestErr error
	var ingestWG sync.WaitGroup
	ingestWG.Add(1)
	go func() {
		defer ingestWG.Done()
		for round := 1; ; round++ {
			select {
			case <-stop:
				return
			default:
			}
			if err := simulateIngest(ctx, notice, round); err != nil {
				ingestErr = err
				return
			}
			atomic.AddInt64(&ingestRounds, 1)
		}
	}()

	// EIGHT racing promotions, each in its own transaction taking FOR UPDATE on the draft row, which is
	// exactly what the service does. Doing it in SQL here rather than through the service keeps the test
	// about the LOCK rather than about handler wiring.
	const racers = 8
	var wg sync.WaitGroup
	results := make([]error, racers)
	start := make(chan struct{})
	for i := 0; i < racers; i++ {
		wg.Add(1)
		go func(idx int) {
			defer wg.Done()
			<-start
			tx, err := migrator.Begin(ctx)
			if err != nil {
				results[idx] = err
				return
			}
			defer func() { _ = tx.Rollback(ctx) }()

			var status string
			// THE LOCK. Every racer queues here.
			if err := tx.QueryRow(ctx,
				`SELECT review_status FROM requirement_drafts WHERE id=$1 AND organization_id=$2 FOR UPDATE`,
				draft, org).Scan(&status); err != nil {
				results[idx] = err
				return
			}
			if status == "approved" {
				results[idx] = tx.Commit(ctx) // a loser: it read the winner's result and changes nothing
				return
			}
			reqID := uuid.New()
			if _, err := tx.Exec(ctx, `
				INSERT INTO requirements (organization_id, id, title, required_level, required_years)
				VALUES ($1,$2,'Cyber Systems Operator','ts_sci',3)`, org, reqID); err != nil {
				results[idx] = err
				return
			}
			if _, err := tx.Exec(ctx, `
				UPDATE requirement_drafts
				   SET review_status='approved', approved_requirement_id=$3, reviewed_by_user_id=$4, reviewed_at=now()
				 WHERE id=$1 AND organization_id=$2`, draft, org, reqID, reviewerID); err != nil {
				results[idx] = err
				return
			}
			results[idx] = tx.Commit(ctx)
		}(i)
	}
	// WARM UP THE INGEST FIRST. The race finished in under half a second on the first attempt and only three
	// ingest rounds landed, so the overlap guard correctly refused to count it. Waiting until the loop is
	// demonstrably running before releasing the racers makes the interleaving real rather than hoped for,
	// which is the whole point of the test.
	deadline := time.Now().Add(2 * time.Second)
	for atomic.LoadInt64(&ingestRounds) < 20 && time.Now().Before(deadline) {
		time.Sleep(5 * time.Millisecond)
	}
	close(start)
	wg.Wait()
	close(stop)
	ingestWG.Wait()

	if ingestErr != nil {
		t.Fatalf("the ingest loop itself failed, so the race proved nothing: %v", ingestErr)
	}
	rounds := atomic.LoadInt64(&ingestRounds)
	if rounds < 25 {
		t.Fatalf("the ingest loop only completed %d rounds, which is too few to have genuinely overlapped "+
			"the promotions; the race proved nothing", rounds)
	}
	t.Logf("ingest rewrote the notice %d times, overlapping the race throughout", rounds)

	// EVERY racer must have completed cleanly. Without this a racer that errored would leave zero
	// requirements and the count assertion below would report CORRUPTION for what is really a broken test.
	// That is exactly what happened on the first run, and a silent racer error is the same class of defect
	// as a discarded Exec error.
	for i, err := range results {
		if err != nil {
			t.Fatalf("racer %d failed, so the race proved nothing: %v", i, err)
		}
	}

	// EXACTLY ONE requirement.
	var reqCount int
	if err := migrator.QueryRow(ctx, `SELECT count(*) FROM requirements WHERE organization_id=$1`, org).Scan(&reqCount); err != nil {
		t.Fatalf("count requirements: %v", err)
	}
	if reqCount != 1 {
		t.Fatalf("CORRUPTION: %d requirements created by %d racing promotions during ingest, want exactly 1", reqCount, racers)
	}

	// The draft is approved exactly once, carries the requirement, and NAMES A HUMAN.
	var status string
	var reqID, reviewer *uuid.UUID
	if err := migrator.QueryRow(ctx,
		`SELECT review_status, approved_requirement_id, reviewed_by_user_id FROM requirement_drafts WHERE id=$1`,
		draft).Scan(&status, &reqID, &reviewer); err != nil {
		t.Fatalf("read draft: %v", err)
	}
	if status != "approved" || reqID == nil || reviewer == nil {
		t.Fatalf("draft state is inconsistent after the race: status=%s requirement=%v reviewer=%v", status, reqID, reviewer)
	}
	t.Cleanup(func() {
		_, _ = migrator.Exec(context.Background(), `DELETE FROM requirements WHERE organization_id=$1`, org)
	})
}

// THE OTHER HALF OF THE GATE ITEM: no ingest path may rewrite an APPROVED requirement. The public cache
// refresh touches sam_opportunities and opportunity_versions; the requirement it produced is tenant
// authored and must survive byte for byte.
func TestIngestNeverRewritesAnApprovedRequirement(t *testing.T) {
	if migrator == nil {
		t.Skip("integration pools not configured")
	}
	ctx := context.Background()
	org, opp, _ := seedPromotionFixture(t)

	var notice string
	if err := migrator.QueryRow(ctx, `SELECT sam_notice_id FROM tracked_opportunities WHERE id=$1`, opp).Scan(&notice); err != nil {
		t.Fatalf("read notice: %v", err)
	}

	reqID := uuid.New()
	if _, err := migrator.Exec(ctx, `
		INSERT INTO requirements (organization_id, id, title, required_level, required_years)
		VALUES ($1,$2,'Approved And Immutable','ts_sci',5)`, org, reqID); err != nil {
		t.Fatalf("seed approved requirement: %v", err)
	}
	t.Cleanup(func() {
		_, _ = migrator.Exec(context.Background(), `DELETE FROM requirements WHERE organization_id=$1`, org)
	})

	// A byte for byte fingerprint of the tenant authored row BEFORE any ingest runs.
	var before string
	fingerprint := `SELECT md5(coalesce(title,'')||coalesce(required_level,'')||coalesce(required_years::text,''))
	                  FROM requirements WHERE id=$1`
	if err := migrator.QueryRow(ctx, fingerprint, reqID).Scan(&before); err != nil {
		t.Fatalf("fingerprint before: %v", err)
	}

	// Twenty ingest rounds, the same thing sam-sync does on its timer.
	for round := 1; round <= 20; round++ {
		if err := simulateIngest(ctx, notice, round); err != nil {
			t.Fatalf("ingest round %d: %v", round, err)
		}
	}

	var after string
	if err := migrator.QueryRow(ctx, fingerprint, reqID).Scan(&after); err != nil {
		t.Fatalf("fingerprint after: %v", err)
	}
	if before != after {
		t.Fatalf("AN INGEST PATH REWROTE AN APPROVED REQUIREMENT. before=%s after=%s", before, after)
	}
	t.Logf("approved requirement survived 20 ingest rounds byte for byte, md5 %s", after)
}

// POSITIVE CONTROL on this suite, required by the continuation protocol. If the fingerprint cannot detect a
// change that definitely happened, the immutability test above proves nothing.
func TestPromotionSuitePositiveControl(t *testing.T) {
	if migrator == nil {
		t.Skip("integration pools not configured")
	}
	ctx := context.Background()
	org := seedOrg(t)
	reqID := uuid.New()
	if _, err := migrator.Exec(ctx, `
		INSERT INTO requirements (organization_id, id, title, required_level, required_years)
		VALUES ($1,$2,'Control Row','secret',1)`, org, reqID); err != nil {
		t.Fatalf("seed: %v", err)
	}
	t.Cleanup(func() {
		_, _ = migrator.Exec(context.Background(), `DELETE FROM requirements WHERE organization_id=$1`, org)
	})

	fingerprint := `SELECT md5(coalesce(title,'')||coalesce(required_level,'')||coalesce(required_years::text,''))
	                  FROM requirements WHERE id=$1`
	var before, after string
	if err := migrator.QueryRow(ctx, fingerprint, reqID).Scan(&before); err != nil {
		t.Fatalf("before: %v", err)
	}
	// A deliberate change. The fingerprint MUST notice.
	if _, err := migrator.Exec(ctx, `UPDATE requirements SET title='Control Row CHANGED' WHERE id=$1`, reqID); err != nil {
		t.Fatalf("mutate: %v", err)
	}
	if err := migrator.QueryRow(ctx, fingerprint, reqID).Scan(&after); err != nil {
		t.Fatalf("after: %v", err)
	}
	if before == after {
		t.Fatal("POSITIVE CONTROL FAILED: the fingerprint did not detect a change that definitely happened, " +
			"so the immutability test above is not testing anything")
	}
}