refactor(scaffold): Phase B language-advantage pass — dichotomy wave
This commit is contained in:
Nick committed
1 parent
5c4ba1cc65
commit
4f9f042452
2 files changed
+302
No files matched your search
@@ -0,0 +1,114 @@
|
||||
package main
|
||||
|
||||
// phaseb.go — Phase B language-advantage pass (dichotomy wave): the oracle
|
||||
// observes repos ONE AT A TIME (each repo_state_from_git call is two
|
||||
// blocking subprocess.run calls). Go's cheap goroutines make the IO fan
|
||||
// out: one goroutine per repo over the SAME GitRunner seam, results merged
|
||||
// by a single writer in canonical rid order — the merged doc's byte form
|
||||
// is EXACTLY the sequential observe_repo sequence's (stamps pinned by the
|
||||
// injected clock), while the git wall-clock cost becomes
|
||||
// max-per-repo instead of the sum. Doc mutations stay on the calling
|
||||
// goroutine (concurrency safety by ownership).
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"sort"
|
||||
"strings"
|
||||
"sync"
|
||||
)
|
||||
|
||||
// RepoSpec — one repo to refresh: the rid is the repo identity's store id
|
||||
// (owner/name); RepoRoot is the working tree the git runner reads.
|
||||
type RepoSpec struct {
|
||||
Owner string
|
||||
Name string
|
||||
RepoRoot string
|
||||
}
|
||||
|
||||
// RepoObservation — the per-repo outcome (degraded observations surface as
|
||||
// the oracle would leave them: head "" and clean nil).
|
||||
type RepoObservation struct {
|
||||
Repo string `json:"repo"`
|
||||
Head string `json:"head"`
|
||||
Clean *bool `json:"clean"`
|
||||
}
|
||||
|
||||
// ObserveReposConcurrently — fan the git IO out (one goroutine per spec,
|
||||
// bounded by `workers`), then re-observe the store SEQUENTIALLY in
|
||||
// canonical rid order so the doc mutations are byte-identical to a
|
||||
// sequential repo_state_from_git walk. Duplicate rids fan out once; the
|
||||
// observation stamps are shared (one injected clock).
|
||||
func ObserveReposConcurrently(doc map[string]interface{}, specs []RepoSpec, workers int, nowStr string) ([]RepoObservation, error) {
|
||||
if workers < 1 {
|
||||
workers = 1
|
||||
}
|
||||
// dedupe by rid (duplicate specs would double-append observations —
|
||||
// the sequential semantics observe each repo once per pass) and
|
||||
// canonicalize the mutation order.
|
||||
canon := make([]RepoSpec, 0, len(specs))
|
||||
seen := map[string]bool{}
|
||||
for _, s := range specs {
|
||||
rid := s.Owner + "/" + s.Name
|
||||
if seen[rid] {
|
||||
continue
|
||||
}
|
||||
seen[rid] = true
|
||||
canon = append(canon, s)
|
||||
}
|
||||
sort.Slice(canon, func(i, j int) bool {
|
||||
return canon[i].Owner+"/"+canon[i].Name < canon[j].Owner+"/"+canon[j].Name
|
||||
})
|
||||
|
||||
type outcome struct {
|
||||
idx int
|
||||
head string
|
||||
clean *bool
|
||||
err error
|
||||
}
|
||||
results := make([]outcome, len(canon))
|
||||
sem := make(chan struct{}, workers)
|
||||
var wg sync.WaitGroup
|
||||
var firstErr error
|
||||
var errMu sync.Mutex
|
||||
for i, s := range canon {
|
||||
if GitRunner == nil {
|
||||
continue
|
||||
}
|
||||
wg.Add(1)
|
||||
go func(i int, root string) {
|
||||
defer wg.Done()
|
||||
sem <- struct{}{}
|
||||
defer func() { <-sem }()
|
||||
head := ""
|
||||
var clean *bool
|
||||
if out, _, err := GitRunner([]string{"git", "-C", root, "rev-parse", "HEAD"}, 15); err == nil {
|
||||
head = strings.TrimSpace(out)
|
||||
if stOut, stCode, err2 := GitRunner([]string{"git", "-C", root, "status", "--short"}, 15); err2 == nil {
|
||||
c := stCode == 0 && strings.TrimSpace(stOut) == ""
|
||||
clean = &c
|
||||
}
|
||||
}
|
||||
results[i] = outcome{idx: i, head: head, clean: clean}
|
||||
}(i, s.RepoRoot)
|
||||
}
|
||||
wg.Wait()
|
||||
|
||||
observed := make([]RepoObservation, 0, len(canon))
|
||||
for i, s := range canon {
|
||||
rid := s.Owner + "/" + s.Name
|
||||
if _, err := ObserveRepo(doc, s.Owner, s.Name, results[i].head, results[i].clean, "", nowStr); err != nil {
|
||||
errMu.Lock()
|
||||
if firstErr == nil {
|
||||
firstErr = fmt.Errorf("observe %s: %w", rid, err)
|
||||
}
|
||||
errMu.Unlock()
|
||||
continue
|
||||
}
|
||||
observed = append(observed, RepoObservation{
|
||||
Repo: rid,
|
||||
Head: results[i].head,
|
||||
Clean: results[i].clean,
|
||||
})
|
||||
}
|
||||
return observed, firstErr
|
||||
}
|
||||
@@ -0,0 +1,188 @@
|
||||
package main
|
||||
|
||||
// phaseb_test.go — Phase B language-advantage pass (dichotomy wave):
|
||||
// determinism of the concurrent observation merge (identical to the
|
||||
// sequential oracle walk), graceful degradation parity, and a -race
|
||||
// hammer over the shared doc.
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"sync"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// sequentialRunner — a deterministic fake git: head derives from the root
|
||||
// (repeatable), status reflects the root name ("dirty" → non-empty).
|
||||
func sequentialRunner(root string) (string, int, error) {
|
||||
if containsSubstr(root, "bad") {
|
||||
return "fatal: not a git repository", 128, nil
|
||||
}
|
||||
return "00000000000112233445566778899aabbccdde" + root, 0, nil
|
||||
}
|
||||
|
||||
func containsSubstr(s, sub string) bool {
|
||||
return len(s) >= len(sub) && (func() bool {
|
||||
for i := 0; i+len(sub) <= len(s); i++ {
|
||||
if s[i:i+len(sub)] == sub {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
})()
|
||||
}
|
||||
|
||||
func TestPhaseBConcurrentMergesLikeSequential(t *testing.T) {
|
||||
specs := []RepoSpec{
|
||||
{"alpha", "one", "r-c"},
|
||||
{"alpha", "two", "r-dirty"},
|
||||
{"beta", "three", "r-c"},
|
||||
}
|
||||
bind := func(doc map[string]interface{}) map[string]interface{} {
|
||||
for _, s := range specs {
|
||||
dup := map[string]bool{}
|
||||
rid := s.Owner + "/" + s.Name
|
||||
if s.Owner != "" && !dup[rid] {
|
||||
dup[rid] = true
|
||||
}
|
||||
if _, err := BindRepo(doc, RepoBinding{
|
||||
Forge: "gitea", Owner: s.Owner, Name: s.Name,
|
||||
DefaultBranch: "m", RepoRoot: s.RepoRoot,
|
||||
}, "2026-10-13T10:50:00Z"); err != nil {
|
||||
t.Fatalf("bind %s: %v", s.Name, err)
|
||||
}
|
||||
}
|
||||
return doc
|
||||
}
|
||||
|
||||
fakeGit := func(args []string, secs int) (string, int, error) {
|
||||
if len(args) > 2 && args[2] == "rev-parse" {
|
||||
head, rc, err := sequentialRunner(args[3])
|
||||
return head, rc, err
|
||||
}
|
||||
if containsSubstr(args[3], "dirty") {
|
||||
return " M x\n", 0, nil
|
||||
}
|
||||
return "", 0, nil
|
||||
}
|
||||
|
||||
// The sequential reference walks the DEDUPED canonical order — the
|
||||
// semantics the concurrent pass promises (each repo observed once per
|
||||
// pass, mutations in canonical rid order).
|
||||
sequential := func() map[string]interface{} {
|
||||
doc := bind(freshGapDoc())
|
||||
prev := GitRunner
|
||||
GitRunner = fakeGit
|
||||
defer func() { GitRunner = prev }()
|
||||
dup := map[string]bool{}
|
||||
for _, s := range specs {
|
||||
rid := s.Owner + "/" + s.Name
|
||||
if dup[rid] {
|
||||
continue
|
||||
}
|
||||
dup[rid] = true
|
||||
if _, err := RepoStateFromGit(doc, s.Owner, s.Name, s.RepoRoot, "2026-10-13T10:50:00Z"); err != nil {
|
||||
t.Fatalf("sequential %v", err)
|
||||
}
|
||||
}
|
||||
return doc
|
||||
}()
|
||||
|
||||
concurrent := func() map[string]interface{} {
|
||||
doc := bind(freshGapDoc())
|
||||
prev := GitRunner
|
||||
GitRunner = fakeGit
|
||||
defer func() { GitRunner = prev }()
|
||||
obs, err := ObserveReposConcurrently(doc, append(append([]RepoSpec{}, specs...),
|
||||
RepoSpec{"alpha", "one", "r-c"}), 8, "2026-10-13T10:50:00Z") // duplicate rid fans out once
|
||||
if err != nil {
|
||||
t.Fatalf("concurrent: %v", err)
|
||||
}
|
||||
if len(obs) != 3 {
|
||||
t.Fatalf("observations: %d (want 3 — duplicate rid fanned once)", len(obs))
|
||||
}
|
||||
return doc
|
||||
}()
|
||||
|
||||
_, a := canonJSON(sequential)
|
||||
_, b := canonJSON(concurrent)
|
||||
if a != b {
|
||||
t.Fatalf("concurrent merge diverges from the sequential walk:\n%s\nvs\n%s", a, b)
|
||||
}
|
||||
}
|
||||
|
||||
func TestPhaseBDegradedObservation(t *testing.T) {
|
||||
doc := freshGapDoc()
|
||||
_, err := BindRepo(doc, RepoBinding{
|
||||
Forge: "gitea", Owner: "o", Name: "r",
|
||||
DefaultBranch: "m", RepoRoot: "x",
|
||||
}, "2026-10-13T10:50:00Z")
|
||||
if err != nil {
|
||||
t.Fatalf("bind: %v", err)
|
||||
}
|
||||
prev := GitRunner
|
||||
GitRunner = func(args []string, secs int) (string, int, error) {
|
||||
return "", 0, errStatic("simulated oserror")
|
||||
}
|
||||
obs, err := ObserveReposConcurrently(doc, []RepoSpec{{"o", "r", "gone"}}, 4, "2026-10-13T10:50:00Z")
|
||||
GitRunner = prev
|
||||
if err != nil {
|
||||
t.Fatalf("degraded observation must not fail: %v", err)
|
||||
}
|
||||
e := EntityWithAliases(doc, "repo:o/r")
|
||||
_ = e
|
||||
if len(obs) != 1 || obs[0].Head != "" || obs[0].Clean != nil {
|
||||
t.Fatalf("degraded outcome: %+v", obs)
|
||||
}
|
||||
}
|
||||
|
||||
// TestPhaseBRaceHammer — interleaves concurrent observation fan-out with
|
||||
// single-repo observes and reconciliation on one shared doc (16 workers);
|
||||
// -race is the gate.
|
||||
func TestPhaseBRaceHammer(t *testing.T) {
|
||||
doc := freshGapDoc()
|
||||
if _, err := BindRepo(doc, RepoBinding{
|
||||
Forge: "gitea", Owner: "h", Name: "r",
|
||||
DefaultBranch: "m", RepoRoot: "x",
|
||||
}, "2026-10-13T10:50:00Z"); err != nil {
|
||||
t.Fatalf("bind: %v", err)
|
||||
}
|
||||
prev := GitRunner
|
||||
GitRunner = func(args []string, secs int) (string, int, error) {
|
||||
return fmt.Sprintf("%x", len(args)) + "0123456789abcdef", 0, nil
|
||||
}
|
||||
defer func() { GitRunner = prev }()
|
||||
|
||||
specs := []RepoSpec{{"h", "r", "x"}, {"h", "r2", "y"}, {"h", "r3", "z"}}
|
||||
for _, s := range specs {
|
||||
if _, err := BindRepo(doc, RepoBinding{
|
||||
Forge: "gitea", Owner: s.Owner, Name: s.Name,
|
||||
DefaultBranch: "m", RepoRoot: s.RepoRoot,
|
||||
}, "2026-10-13T10:50:00Z"); err != nil {
|
||||
t.Fatalf("bind %s: %v", s.Name, err)
|
||||
}
|
||||
}
|
||||
var wg sync.WaitGroup
|
||||
for w := 0; w < 16; w++ {
|
||||
wg.Add(1)
|
||||
go func(w int) {
|
||||
defer wg.Done()
|
||||
for i := 0; i < 5; i++ {
|
||||
var err error
|
||||
switch w % 3 {
|
||||
case 0:
|
||||
_, err = RepoStateFromGit(doc, "h", "r", "x", "2026-10-13T10:50:00Z")
|
||||
case 1:
|
||||
_, err = ObserveRepo(doc, "h", "r", "aaa", nil, "", "2026-10-13T10:50:00Z")
|
||||
case 2:
|
||||
_, err = ObserveReposConcurrently(doc, specs, 3, "2026-10-13T10:50:00Z")
|
||||
if err != nil {
|
||||
t.Errorf("hammer concurrent: %v", err)
|
||||
}
|
||||
}
|
||||
_ = err
|
||||
Reconcile(doc, "2026-10-13T10:50:00Z")
|
||||
}
|
||||
}(w)
|
||||
}
|
||||
wg.Wait()
|
||||
}
|
||||
Reference in new issue
Block a user