Janady13/missionmatch / tools/verify/stress/main.go

Source identity

Field	Recorded value
Source type	GitHub source
Original record	Janady13/missionmatch / tools/verify/stress/main.go
Date	Not specified in source
Repository	Janady13/missionmatch
Revision	720d47fae476b0d38477ae4401697d372b1ea2bc
Branch	feature/opportunity-to-outcome
Path or archive member	tools/verify/stress/main.go
Original SHA-256	d185c0ec17b13ff007470643e9507bf60562d7eb08afaa198d9eaa4865d3b5ce
Git blob SHA	e91e443ad6a9b4c19d73acbbbb5416119de06a55
Reading-copy SHA-256	d185c0ec17b13ff007470643e9507bf60562d7eb08afaa198d9eaa4865d3b5ce
Original source ↗
Recovered source transcript

Download complete source text ↓298 lines · 11.1 KB
// Command stress is the Gate B instrument: it hurts the system on purpose and reports numbers.
//
// Licensed by docs/BUILD_CONTINUATION_PROTOCOL.md Section 0 item 4 as tools/verify/stress. It runs against
// STAGING, which carries 2,766 organizations and is the right place to break things.
//
// EVERY PROFILE HAS A POSITIVE CONTROL, and the control runs FIRST. The protocol requires it because an
// instrument that passes by doing nothing is worse than no instrument: this build has already been burned
// four separate times by a check that returned a clean negative while measuring nothing. So each profile
// can be asked to plant a defect and must FAIL on it before its real run is allowed to count.
//
//	stress -profile race    -control    # must FAIL, proving the race detector detects
//	stress -profile race                # the counted run
//
// Usage:
//
//	stress -dsn <migrator dsn> -profile race|load|soak [-n 8] [-control]
package main

import (
	"context"
	"flag"
	"fmt"
	"log"
	"os"
	"sort"
	"sync"
	"sync/atomic"
	"time"

	"github.com/google/uuid"
	"github.com/jackc/pgx/v5/pgxpool"
)

func main() {
	dsn := flag.String("dsn", os.Getenv("MM_MIGRATOR_DSN"), "migrator DSN to the staging database")
	profile := flag.String("profile", "race", "race, load, soak, volume, concurrent-extract, scrub-flood, or http-load")
	n := flag.Int("n", 8, "concurrency for the race profile")
	control := flag.Bool("control", false, "plant a defect: the profile MUST fail, proving it can detect")
	dur := flag.Duration("duration", 30*time.Second, "duration for load and soak")
	base := flag.String("base", "", "base url for the http-load profile, for example https://host")
	roles := flag.String("roles", "", "http-load logins as email:password, comma separated")
	targetRPS := flag.Int("rps", 50, "http-load target requests per second, the protocol asks for 50 sustained")
	flag.Parse()

	// THE PURE PROFILES NEED NO DATABASE and are deliberately not given one. volume, concurrent-extract and
	// scrub-flood measure CPU, allocation, and the egress gate, none of which involve Postgres. Demanding a
	// DSN for them would be a reason to skip them, and skipping them is how three of the five Phase 2 stress
	// items went unrun while the gate reported PASS.
	switch *profile {
	case "volume", "concurrent-extract", "scrub-flood", "http-load":
		log.Printf("stress: profile=%s control=%v, no database required", *profile, *control)
		var pureErr error
		switch *profile {
		case "volume":
			pureErr = volumeProfile(*control)
		case "concurrent-extract":
			pureErr = concurrentExtractProfile(*control)
		case "scrub-flood":
			pureErr = scrubFloodProfile(*control)
		case "http-load":
			pureErr = httpLoadProfile(*base, *roles, *dur, *targetRPS, *control)
		}
		if *control {
			if pureErr == nil {
				log.Fatalf("POSITIVE CONTROL FAILED: profile %s passed with a planted defect, so it cannot "+
					"detect one and its real run would prove nothing", *profile)
			}
			log.Printf("positive control OK: the planted defect was DETECTED (%v)", pureErr)
			return
		}
		if pureErr != nil {
			log.Fatalf("STRESS FAILED: %v", pureErr)
		}
		log.Printf("stress: profile %s PASSED", *profile)
		return
	}

	if *dsn == "" {
		log.Fatal("stress: -dsn or MM_MIGRATOR_DSN is required")
	}
	ctx := context.Background()
	cfg, err := pgxpool.ParseConfig(*dsn)
	if err != nil {
		log.Fatalf("stress: dsn: %v", err)
	}
	// Deliberately larger than any racer count used here: a pool smaller than the concurrency deadlocks and
	// looks exactly like the contention bug the race profile exists to find.
	cfg.MaxConns = 64
	pool, err := pgxpool.NewWithConfig(ctx, cfg)
	if err != nil {
		log.Fatalf("stress: pool: %v", err)
	}
	defer pool.Close()

	var db string
	if err := pool.QueryRow(ctx, "select current_database()").Scan(&db); err != nil {
		log.Fatalf("stress: connect: %v", err)
	}
	if db == "missionmatch" {
		log.Fatal("stress: REFUSING to run against a database named missionmatch. This tool creates and " +
			"destroys rows and belongs on staging only.")
	}
	log.Printf("stress: profile=%s target=%s concurrency=%d control=%v", *profile, db, *n, *control)

	var perr error
	switch *profile {
	case "race":
		perr = raceProfile(ctx, pool, *n, *control)
	case "load", "soak":
		perr = loadProfile(ctx, pool, *profile, *dur, *control)
	default:
		log.Fatalf("stress: unknown profile %q", *profile)
	}

	if *control {
		if perr == nil {
			log.Fatalf("POSITIVE CONTROL FAILED: profile %s passed with a planted defect, so it cannot "+
				"detect one and its real run would prove nothing", *profile)
		}
		log.Printf("positive control OK: the planted defect was DETECTED (%v)", perr)
		return
	}
	if perr != nil {
		log.Fatalf("STRESS FAILED: %v", perr)
	}
	log.Printf("stress: profile %s PASSED", *profile)
}

// raceProfile races N concurrent claims of ONE row through the same FOR UPDATE discipline the recruiting
// and promotion paths use. Exactly one must win.
//
// The planted defect for the control drops the FOR UPDATE, which is precisely the mistake that would make
// every lock in this system decorative, so the control proves the detector notices the thing that matters.
func raceProfile(ctx context.Context, pool *pgxpool.Pool, n int, control bool) error {
	org := uuid.New()
	if _, err := pool.Exec(ctx,
		`INSERT INTO organizations (id, name, slug) VALUES ($1,$2,$3)`,
		org, "stress-"+org.String()[:8], "stress-"+org.String()[:8]); err != nil {
		return fmt.Errorf("seed org: %w", err)
	}
	defer func() {
		_, _ = pool.Exec(context.Background(), `DELETE FROM requirements WHERE organization_id=$1`, org)
		_, _ = pool.Exec(context.Background(), `DELETE FROM organizations WHERE id=$1`, org)
	}()

	seat := uuid.New()
	if _, err := pool.Exec(ctx,
		`INSERT INTO requirements (organization_id, id, title, required_level, required_years)
		 VALUES ($1,$2,'stress seat','secret',1)`, org, seat); err != nil {
		return fmt.Errorf("seed seat: %w", err)
	}

	// THE PLANTED DEFECT, and it has to be DETERMINISTIC. The first attempt simply dropped FOR UPDATE and
	// still produced one winner, because the timing did not happen to interleave: that proved the harness
	// ran, not that the assertion can catch a second winner. A control that only sometimes manifests is
	// itself an instrument that passes by doing nothing, which is the whole failure mode being guarded
	// against. So the control removes the ALREADY CLAIMED GUARD, which makes every racer claim and
	// guarantees the "exactly one winner" assertion is put under real pressure.
	lock := "FOR UPDATE"
	skipGuard := control

	var winners int64
	var wg sync.WaitGroup
	start := make(chan struct{})
	for i := 0; i < n; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			<-start
			tx, err := pool.Begin(ctx)
			if err != nil {
				return
			}
			defer func() { _ = tx.Rollback(ctx) }()
			var title string
			if err := tx.QueryRow(ctx,
				`SELECT title FROM requirements WHERE id=$1 AND organization_id=$2 `+lock, seat, org).Scan(&title); err != nil {
				return
			}
			if title != "stress seat" && !skipGuard {
				_ = tx.Commit(ctx) // already claimed by the winner
				return
			}
			// Without the lock, several racers read "stress seat" before any of them writes, and several
			// claim. That is the corruption this profile exists to catch.
			time.Sleep(3 * time.Millisecond)
			if _, err := tx.Exec(ctx,
				`UPDATE requirements SET title='claimed' WHERE id=$1 AND organization_id=$2`, seat, org); err != nil {
				return
			}
			if err := tx.Commit(ctx); err == nil {
				atomic.AddInt64(&winners, 1)
			}
		}()
	}
	close(start)
	wg.Wait()

	got := atomic.LoadInt64(&winners)
	log.Printf("  race n=%d winners=%d (want exactly 1)", n, got)
	if got != 1 {
		return fmt.Errorf("race at n=%d elected %d winners, want exactly 1", n, got)
	}
	return nil
}

// loadProfile drives a representative read against the tenant plane and reports latency percentiles. It
// measures the DATABASE path rather than HTTP, because the authenticated HTTP profile needs seeded fixture
// sessions per role and that is Gate C's fixture work; this is the part that can be measured today without
// pretending otherwise.
//
// The planted defect for the control is an absurd latency target, so the control proves the profile
// actually compares against its target rather than always reporting success.
func loadProfile(ctx context.Context, pool *pgxpool.Pool, name string, dur time.Duration, control bool) error {
	target := 300 * time.Millisecond // p95 target from the protocol
	if control {
		target = 1 * time.Nanosecond // THE PLANTED DEFECT: unmeetable, so a passing run would be a lie.
	}
	if name == "soak" {
		dur = dur * 2
	}

	deadline := time.Now().Add(dur)
	var lats []time.Duration
	var errs int64
	var wg sync.WaitGroup
	for w := 0; w < 8; w++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			var local []time.Duration
			for time.Now().Before(deadline) {
				t0 := time.Now()
				var c int
				if err := pool.QueryRow(ctx, `SELECT count(*) FROM organizations`).Scan(&c); err != nil {
					atomic.AddInt64(&errs, 1)
					continue
				}
				local = append(local, time.Since(t0))
			}
			mu.Lock()
			lats = append(lats, local...)
			mu.Unlock()
		}()
	}
	wg.Wait()

	if len(lats) == 0 {
		return fmt.Errorf("%s produced no samples", name)
	}
	p95 := percentile(lats, 95)
	p99 := percentile(lats, 99)
	rps := float64(len(lats)) / dur.Seconds()
	log.Printf("  %s samples=%d rps=%.0f p95=%s p99=%s errors=%d target_p95=%s",
		name, len(lats), rps, p95.Round(time.Microsecond), p99.Round(time.Microsecond), errs, target)
	if errs > 0 {
		return fmt.Errorf("%s produced %d errors, target is zero", name, errs)
	}
	if p95 > target {
		return fmt.Errorf("%s p95 %s exceeds the %s target", name, p95, target)
	}
	return nil
}

var mu sync.Mutex

func percentile(d []time.Duration, p int) time.Duration {
	if len(d) == 0 {
		return 0
	}
	// FINDING D16. This used an INSERTION SORT, with a comment saying "sorting a few thousand durations is not
	// the bottleneck here and a simple implementation is easier to trust than a clever one in a measuring
	// instrument". The reasoning was sound and the premise was wrong.
	//
	// A soak at the duration THE PROTOCOL ACTUALLY ASKS FOR does not produce a few thousand samples. Thirty
	// minutes at roughly a thousand reads per second produces about 1.8 MILLION, and insertion sort is O(n
	// squared), so it needs on the order of a trillion operations. Called twice, once for p95 and once for p99.
	//
	// THE CONSEQUENCE IS THAT THE PROTOCOL'S SOAK HAS NEVER BEEN MEASURABLE WITH THIS TOOL. The 60 second soak
	// in the first Gate B bundle completed because 60,000 samples is still tractable; the 30 minute soak the
	// protocol requires hangs after the load has finished, burning CPU in the reporting step with every number
	// already collected and none of it printable. Two long soaks were lost to this before it was diagnosed, and
	// the first loss was wrongly blamed on an unrelated credential rotation happening at the same time.
	//
	// A measuring instrument that cannot report the measurement the protocol asks for is not a slow instrument,
	// it is a missing one.
	c := append([]time.Duration(nil), d...)
	sort.Slice(c, func(i, j int) bool { return c[i] < c[j] })
	idx := (len(c)*p)/100 - 1
	if idx < 0 {
		idx = 0
	}
	if idx >= len(c) {
		idx = len(c) - 1
	}
	return c[idx]
}