Janady13/missionmatch / internal/aioe/lockguard_test.go

Source identity

Field	Recorded value
Source type	GitHub source
Original record	Janady13/missionmatch / internal/aioe/lockguard_test.go
Date	Not specified in source
Repository	Janady13/missionmatch
Revision	720d47fae476b0d38477ae4401697d372b1ea2bc
Branch	feature/opportunity-to-outcome
Path or archive member	internal/aioe/lockguard_test.go
Original SHA-256	dedc8be98d909c1a6fcf5ce413f9aa1504e4216c4fb7d094b4708de5a2ac1436
Git blob SHA	0bf3604112df7f9f480ee2354e8ec659128f550c
Reading-copy SHA-256	dedc8be98d909c1a6fcf5ce413f9aa1504e4216c4fb7d094b4708de5a2ac1436

Original source ↗

Every recorded occurrence

These 2 records contain identical original source bytes. Their paths and revisions remain part of the archive.

Original record	Branch or archive	Revision / identifier	Original location
Janady13/missionmatch / internal/aioe/lockguard_test.go	feature/opportunity-to-outcome	720d47fae476b0d38477ae4401697d372b1ea2bc	Open original
Janady13/missionmatch / internal/aioe/lockguard_test.go	feature/aioe-upgrade	2178e7298d0e5d9af83afa12402fd587f0ba6ed8	Open original

Recovered source transcript

Download complete source text ↓121 lines · 4.9 KB

package aioe

import (
	"go/ast"
	"go/parser"
	"go/token"
	"os"
	"strings"
	"testing"
)

// funcBody returns the source text of one top level method body in updater.go. The lost update this file
// guards is a property of WHERE a statement sits relative to the transaction, not of any value the code
// computes, so it cannot be observed from the package API without a live Postgres and two racing
// connections. Reading the shape of the source is therefore the honest unit level proof, in the same
// spirit as sqlguard_test.go, which proves an isolation property the same way.
func funcBody(t *testing.T, file, name string) string {
	t.Helper()
	src, err := os.ReadFile(file)
	if err != nil {
		t.Fatal(err)
	}
	fset := token.NewFileSet()
	parsed, err := parser.ParseFile(fset, file, src, 0)
	if err != nil {
		t.Fatal(err)
	}
	for _, decl := range parsed.Decls {
		fn, ok := decl.(*ast.FuncDecl)
		if !ok || fn.Name.Name != name || fn.Body == nil {
			continue
		}
		start := fset.Position(fn.Body.Pos()).Offset
		end := fset.Position(fn.Body.End()).Offset
		return string(src[start:end])
	}
	t.Fatalf("%s: function %s not found", file, name)
	return ""
}

// stripComments removes // comment text so an assertion about the SQL cannot be satisfied by prose that
// merely mentions FOR UPDATE.
func stripComments(body string) string {
	var out strings.Builder
	for _, line := range strings.Split(body, "\n") {
		if strings.HasPrefix(strings.TrimSpace(line), "//") {
			continue
		}
		out.WriteString(line)
		out.WriteString("\n")
	}
	return out.String()
}

// TestRunOrgReadsActiveProfileInsideTheLockedTransaction pins the lost update fix. The active weight
// profile MUST be read on the transaction that writes the new profile, and that read MUST take a row lock.
// Read on the raw pool before the transaction opens, a Promote committing in between silently reverts the
// promoted weights, and the aioe_weight_adjustments trail records an old_value from a profile that was no
// longer in effect, which is a false compliance record for an automated employment decision tool.
func TestRunOrgReadsActiveProfileInsideTheLockedTransaction(t *testing.T) {
	body := stripComments(funcBody(t, "updater.go", "runOrg"))

	const activeRead = "FROM aioe_weight_profiles"
	readAt := strings.Index(body, activeRead)
	if readAt < 0 {
		t.Fatal("runOrg no longer reads the active profile; this guard needs updating with the code")
	}
	beginAt := strings.Index(body, "u.db.Begin(ctx)")
	if beginAt < 0 {
		t.Fatal("runOrg no longer opens its own transaction; this guard needs updating with the code")
	}
	if readAt < beginAt {
		t.Fatal("the active profile is read BEFORE the transaction opens: a concurrent Promote between the read and the write silently reverts the promoted weights and falsifies the adjustment trail")
	}
	// The read must run ON the transaction. A read on u.db from inside the transaction body borrows a
	// DIFFERENT pooled connection, takes no lock the transaction holds, and gains nothing, so the receiver
	// is checked and not merely the ordering.
	callAt := strings.LastIndex(body[:readAt], "QueryRow(")
	if callAt < 3 {
		t.Fatal("the active profile read is not a QueryRow call; this guard needs updating with the code")
	}
	if body[callAt-3:callAt] != "tx." {
		t.Fatalf("the active profile must be read on the transaction (tx.QueryRow), not on the pool, got %q", strings.TrimSpace(body[callAt-8:callAt+9]))
	}
	tail := body[readAt:]
	stop := strings.Index(tail, "`")
	if stop < 0 {
		t.Fatal("could not find the end of the active profile query")
	}
	if !strings.Contains(tail[:stop], "FOR UPDATE") {
		t.Fatal("the active profile read must take a row lock (FOR UPDATE) so the read and the write are atomic against a concurrent Promote")
	}
}

// TestPromoteLocksTheActiveProfileFirst is the other half of the same mutual exclusion. Promote must lock
// the SAME active row runOrg locks, and must take it before it touches any candidate row, so the two
// writers serialize on one row in one order rather than deadlocking on two.
func TestPromoteLocksTheActiveProfileFirst(t *testing.T) {
	body := stripComments(funcBody(t, "updater.go", "Promote"))

	activeAt := strings.Index(body, "status = 'active'")
	if activeAt < 0 {
		t.Fatal("Promote no longer reads the active profile; this guard needs updating with the code")
	}
	candAt := strings.Index(body, "status = 'candidate'")
	if candAt < 0 {
		t.Fatal("Promote no longer reads the candidate profile; this guard needs updating with the code")
	}
	if activeAt > candAt {
		t.Fatal("Promote must lock the active profile BEFORE the candidate, so the shadow pass (which locks active then retires candidates) cannot deadlock against it")
	}
	tail := body[activeAt:]
	stop := strings.Index(tail, "`")
	if stop < 0 {
		t.Fatal("could not find the end of Promote's active profile query")
	}
	if !strings.Contains(tail[:stop], "FOR UPDATE") {
		t.Fatal("Promote must lock the active profile row (FOR UPDATE) so a concurrent nightly pass cannot compute from the profile it is about to supersede")
	}
}