Compare commits
19
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
6a70735489 | ||
|
|
25176e1ccd | ||
|
|
7e1f3de77a | ||
|
|
4b1bed4b9d | ||
|
|
25d69406b7 | ||
|
|
f836c021e1 | ||
|
|
450f0bbe79 | ||
|
|
5a7076a6a2 | ||
|
|
30b22c198e | ||
|
|
753e8fc202 | ||
|
|
e5202445ba | ||
|
|
dcae699323 | ||
|
|
70ef99595a | ||
|
|
746eb885a3 | ||
|
|
2cdb9b8fc4 | ||
|
|
e830045545 | ||
|
|
454fb8a918 | ||
|
|
c32474ee40 | ||
|
|
f95f3f430e |
+28
-9
@@ -83,6 +83,9 @@ export function checkStatusOf(change: OwedChange): { owner: string; repo: string
|
||||
return { owner, repo, sha, revision, status };
|
||||
}
|
||||
|
||||
/** How many mesh/check writes this module has at the forge at once. */
|
||||
export const WRITES_AT_ONCE = 4;
|
||||
|
||||
/** The one writer of mesh/check: each owed revision of the checker's, in revision order per key. */
|
||||
export class CheckStatuses {
|
||||
/** The newest revision written of each key, and the newest seen; the oldest keys go past what any forge holds open. */
|
||||
@@ -91,6 +94,9 @@ export class CheckStatuses {
|
||||
/** The turn each key's writes wait on, and the asks again still running. */
|
||||
private turns = new Map<string, Promise<unknown>>();
|
||||
private again = new Set<Promise<void>>();
|
||||
/** Writes at the forge at once, at most WRITES_AT_ONCE: a watch's start hands over every key together (issue 1726). */
|
||||
private writing = 0;
|
||||
private queued: (() => void)[] = [];
|
||||
private forge: CheckForge;
|
||||
private say: (line: string) => void;
|
||||
private sleep: (ms: number) => Promise<void>;
|
||||
@@ -123,7 +129,7 @@ export class CheckStatuses {
|
||||
this.say(`[gitea] ${asked.owner}/${asked.repo}@${asked.sha.slice(0, 12)}: mesh/check at revision ${asked.revision} is older than ${newest}, already written, and is dropped`);
|
||||
return "dropped";
|
||||
}
|
||||
await this.forge.setCommitStatus(asked.owner, asked.repo, asked.sha, asked.status);
|
||||
await this.atTheForge(() => this.forge.setCommitStatus(asked.owner, asked.repo, asked.sha, asked.status));
|
||||
this.remember(this.written, key, asked.revision);
|
||||
return "written";
|
||||
});
|
||||
@@ -133,19 +139,32 @@ export class CheckStatuses {
|
||||
return turn;
|
||||
}
|
||||
|
||||
/** What the watch hands over: written, or when the forge refuses, asked again on a bound in the background until it is
|
||||
* written or a newer revision of the key was seen. Never throws: a watch's handler that throws loses the change. */
|
||||
/** What the watch hands over: taken at once and written in the background, or when the forge refuses, asked again on a
|
||||
* bound until it is written or a newer revision of the key was seen. Never throws: a watch's handler that throws loses
|
||||
* the change. It never waits on the forge (novox/hq issue 1726): the runtime hands a watch's current values one at a
|
||||
* time, waits for each to be taken and gives the whole set 10 s; a forge write per key made a large owed outlast that
|
||||
* bound at every attempt, so the watch never held and no verdict reached mesh/check. Each key's writes still take
|
||||
* turns in revision order (apply). */
|
||||
async take(change: OwedChange): Promise<void> {
|
||||
const run = this.apply(change).then(() => {}, (err) => this.askAgain(change, err));
|
||||
this.again.add(run);
|
||||
void run.finally(() => this.again.delete(run));
|
||||
}
|
||||
|
||||
/** Runs one forge write when fewer than WRITES_AT_ONCE are running, else after one ends. */
|
||||
private async atTheForge<T>(write: () => Promise<T>): Promise<T> {
|
||||
if (this.writing >= WRITES_AT_ONCE) await new Promise<void>((r) => this.queued.push(r));
|
||||
else this.writing++;
|
||||
try {
|
||||
await this.apply(change);
|
||||
} catch (err) {
|
||||
const run = this.askAgain(change, err);
|
||||
this.again.add(run);
|
||||
void run.finally(() => this.again.delete(run));
|
||||
return await write();
|
||||
} finally {
|
||||
const next = this.queued.shift();
|
||||
if (next) next();
|
||||
else this.writing--;
|
||||
}
|
||||
}
|
||||
|
||||
/** Resolves when no ask again is running. */
|
||||
/** Resolves when no write or ask again is running. */
|
||||
async settled(): Promise<void> {
|
||||
while (this.again.size) await Promise.all([...this.again]);
|
||||
}
|
||||
|
||||
@@ -133,3 +133,45 @@ test("no tool writes mesh/check: gitea_check_status is gone and gitea_commit_sta
|
||||
const index = readFileSync(new URL("../index.ts", import.meta.url), "utf8");
|
||||
assert.ok(!/gitea_check_status|checkStatusTool/.test(index), "index.ts serves no gitea_check_status");
|
||||
});
|
||||
|
||||
test("the watch's current values are taken without waiting on the forge, so a slow forge never times the watch out (issue 1726)", async () => {
|
||||
// The tool runtime hands a watch's current values one at a time and waits for each to be taken; it gives the whole set
|
||||
// 10 s (mesh-tools StateTimeout) and otherwise ends the watch. On 2026-10-23 the owed held ~200 keys, each written to the
|
||||
// forge before the next was handed over: the set took longer than 10 s, the watch ended at every attempt, and no
|
||||
// verdict reached the forge's mesh/check again.
|
||||
const { CheckStatuses, watchOwed } = await import("../check.ts");
|
||||
const sent: string[] = [];
|
||||
let release: () => void = () => {};
|
||||
const slow = new Promise<void>((r) => { release = r; });
|
||||
const f = { setCommitStatus: async (_o: string, _r: string, s: string, st: any) => { await slow; sent.push(`${s.slice(0, 4)} ${st.state}`); } };
|
||||
const checks = new CheckStatuses(f, () => {}, noSleep);
|
||||
const keys = Array.from({ length: 200 }, (_, i) => key(i.toString(16).padStart(4, "0") + sha.slice(4), "novox/mesh-sdk", i + 1));
|
||||
let handedOver = 0;
|
||||
let attempts = 0;
|
||||
const watch = async (handler: (c: any) => Promise<void>) => {
|
||||
if (++attempts > 1) return; // a second attempt is not what this test waits for: the first must hold
|
||||
// The runtime's bound on the current values: the forge has answered none of them yet.
|
||||
const timedOut = new Promise<"timeout">((r) => setTimeout(() => r("timeout"), 200));
|
||||
const delivered = (async () => { for (const [i, k] of keys.entries()) { await handler(put(k, owed(), i + 1, true)); handedOver++; } return "delivered" as const; })();
|
||||
if (await Promise.race([delivered, timedOut]) === "timeout") throw new Error("the current values of mesh-delivery-checker.owed did not arrive in 10s");
|
||||
};
|
||||
const said: string[] = [];
|
||||
await watchOwed(watch, checks, noSleep, (l) => said.push(l));
|
||||
assert.equal(handedOver, 200, "every current value was taken before the forge answered");
|
||||
assert.ok(!said.some((l) => /could not be bound/.test(l)), said.join(" | "));
|
||||
assert.equal(sent.length, 0, "nothing is written before the forge answers");
|
||||
release();
|
||||
await checks.settled();
|
||||
assert.equal(sent.length, 200, "every owed status is written once the forge answers");
|
||||
});
|
||||
|
||||
test("the writes a watch's start hands over go to the forge a few at a time, never all at once (issue 1726)", async () => {
|
||||
const { CheckStatuses, WRITES_AT_ONCE } = await import("../check.ts");
|
||||
let running = 0, most = 0, done = 0;
|
||||
const f = { setCommitStatus: async () => { running++; most = Math.max(most, running); await new Promise((r) => setTimeout(r, 1)); running--; done++; } };
|
||||
const checks = new CheckStatuses(f, () => {}, noSleep);
|
||||
for (let i = 0; i < 50; i++) await checks.take(put(key(i.toString(16).padStart(4, "0") + sha.slice(4), "novox/mesh-sdk", i + 1), owed(), i + 1, true));
|
||||
await checks.settled();
|
||||
assert.equal(done, 50);
|
||||
assert.equal(most, WRITES_AT_ONCE);
|
||||
});
|
||||
|
||||
@@ -100,6 +100,9 @@ func (m *Module) Puller(steps []StepTool, source buildstep.AskSource) *buildstep
|
||||
if err := m.room(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if a.Compose != nil {
|
||||
return m.compose(ctx, ask.Fetch, a)
|
||||
}
|
||||
return buildstep.RunAsk(ctx, ask, a)
|
||||
}}
|
||||
}
|
||||
|
||||
@@ -4,7 +4,6 @@ package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"os"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
@@ -14,16 +13,25 @@ import (
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/buildstep"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"git.novox.be/novox/mesh-sdk/go/testbus"
|
||||
)
|
||||
|
||||
// checkCommit is the commit the check ask fetches, apart from the delivery ask's, so the fetch order names which ran first.
|
||||
const checkCommit = "89abcdef0123456789abcdef0123456789abcdef"
|
||||
|
||||
// 1887/AC-order, 1887/AC-pull against the real bus: a check ask published before a delivery ask runs after it, and at
|
||||
// capacity 1 the second waits in the stream until the first ends.
|
||||
//
|
||||
// **A bus of its own.** The integration level runs every directory's tests at once on one bus, and this test names the
|
||||
// build seat's stream and its fixed workers (SEAT_NODE_BUILD_GO and its workers) and deletes the stream at its end, so
|
||||
// anything else on that bus, a puller or a leftover consumer, could hold or take its delivery ask and let the check
|
||||
// ask run first. It starts from testbus (mesh-sdk, novox/hq ADR 0495 clause 5): a bus for this test alone, gone after it.
|
||||
//
|
||||
// **The order is read from the record.** At capacity 1 the puller starts the check the instant the delivery ask's step
|
||||
// ends, well before any poll, so "the check key holds a state" is true on correct behaviour (issue 1887). The test
|
||||
// waits for both asks to end and compares the step records' own Started and Ended times, and the fetch order by commit.
|
||||
func TestTheModuleTakesADeliveryAskBeforeAnEarlierCheck(t *testing.T) {
|
||||
url := os.Getenv("MESH_TEST_NATS")
|
||||
if url == "" {
|
||||
t.Skip("MESH_TEST_NATS names no bus: the integration level starts one")
|
||||
}
|
||||
nc, err := nats.Connect(url)
|
||||
nc, err := nats.Connect(testbus.URL(t))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
@@ -40,6 +48,7 @@ func TestTheModuleTakesADeliveryAskBeforeAnEarlierCheck(t *testing.T) {
|
||||
t.Cleanup(func() { _ = js.DeleteStream(context.Background(), "SEAT_NODE_BUILD_GO") })
|
||||
check := deliveryAsk()
|
||||
check.Kind = delivery.AskCheck
|
||||
check.Build.Fetch["commit"] = checkCommit
|
||||
first := deliveryAsk()
|
||||
for _, a := range []delivery.StepAsk{check, first} {
|
||||
raw, _ := delivery.EncodeStepAsk(a)
|
||||
@@ -68,30 +77,36 @@ func TestTheModuleTakesADeliveryAskBeforeAnEarlierCheck(t *testing.T) {
|
||||
p := m.Puller(steps, &buildstep.JetStreamAsks{JS: js, Language: delivery.AskGo})
|
||||
p.PullWait = 200 * time.Millisecond
|
||||
go p.Pull(ctx)
|
||||
for _, a := range []delivery.StepAsk{first, check} {
|
||||
key, _ := delivery.AskStepKey("ace", a.ID)
|
||||
for deadline := time.Now().Add(10 * time.Second); ; time.Sleep(10 * time.Millisecond) {
|
||||
states.mu.Lock()
|
||||
v := states.steps[key]
|
||||
states.mu.Unlock()
|
||||
if len(v) > 0 && v[len(v)-1].State != delivery.StepRunning {
|
||||
break
|
||||
}
|
||||
if time.Now().After(deadline) {
|
||||
t.Fatalf("the ask %s (%s) never ended", a.ID, a.Kind)
|
||||
}
|
||||
}
|
||||
var ran map[string][]delivery.Step
|
||||
for deadline := time.Now().Add(10 * time.Second); ; time.Sleep(10 * time.Millisecond) {
|
||||
ran = map[string][]delivery.Step{}
|
||||
ended := 0
|
||||
states.mu.Lock()
|
||||
checkKey, _ := delivery.AskStepKey("ace", check.ID)
|
||||
if a.ID == first.ID && len(states.steps[checkKey]) > 0 {
|
||||
states.mu.Unlock()
|
||||
t.Fatal("the check ask started before the delivery ask ended at capacity 1")
|
||||
for _, a := range []delivery.StepAsk{first, check} {
|
||||
key, _ := delivery.AskStepKey("ace", a.ID)
|
||||
v := append([]delivery.Step(nil), states.steps[key]...)
|
||||
ran[a.ID] = v
|
||||
if len(v) > 0 && v[len(v)-1].State != delivery.StepRunning {
|
||||
ended++
|
||||
}
|
||||
}
|
||||
states.mu.Unlock()
|
||||
if ended == 2 {
|
||||
break
|
||||
}
|
||||
if time.Now().After(deadline) {
|
||||
t.Fatalf("the asks never both ended: %v", ran)
|
||||
}
|
||||
}
|
||||
// The check's step may start the moment the delivery's ends, so the order is read from the records' own times.
|
||||
deliveryEnded := ran[first.ID][len(ran[first.ID])-1].Ended
|
||||
checkStarted := ran[check.ID][0].Started
|
||||
if checkStarted.Before(deliveryEnded) {
|
||||
t.Fatalf("the check ask started at %v, before the delivery ask ended at %v at capacity 1", checkStarted, deliveryEnded)
|
||||
}
|
||||
mu.Lock()
|
||||
defer mu.Unlock()
|
||||
if len(order) != 2 {
|
||||
t.Fatalf("fetched %v", order)
|
||||
if len(order) != 2 || order[0] != askCommit || order[1] != checkCommit {
|
||||
t.Fatalf("fetched %v, want the delivery ask's commit %s first, then the check's %s", order, askCommit, checkCommit)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -174,7 +174,7 @@ func TestACheckAsksDigestIsTheToolsAnswerInTheCall(t *testing.T) {
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if l := out.(delivery.CheckAnswer).Levels; len(l) != 1 || !strings.Contains(string(l[0].Digest), `"x/a":"d1"`) {
|
||||
if l := out.(delivery.CheckAnswer).Levels; len(l) != 1 || l[0].Digest.Map()["x/a"] != "d1" {
|
||||
t.Fatalf("levels = %+v, want the digest's packages", l)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,210 @@
|
||||
package main
|
||||
|
||||
// **A compose ask runs mesh-sdk's composer at a commit** (novox/hq ADR 0482 clause 4, issues 1621 and 1622): the
|
||||
// assigner asks it, to compose a group with its mesh-sdk member's head, and to read the group's manifests with the
|
||||
// mesh-sdk the running controller and node-engine pin (ADR 0482 clause 5, issue 1623). The module fetches the commit
|
||||
// (its fetch step: the forge key and the network, none of the change's code), builds the composer package in the Go
|
||||
// module the ask names, in a run step shut off with nothing fetched, and keeps the program under its key
|
||||
// (delivery.ComposerKey), so each commit is built once on a node. It fills the request's manifests from the delivery
|
||||
// indexer's manifests state, by digest, and runs the program shut off: no network, none of the account, the request a
|
||||
// file of a workspace of its own. The answer is the composer's verdict whole (delivery.ComposeAnswer); the module
|
||||
// judges nothing of it.
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"sync"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/buildstep"
|
||||
"git.novox.be/novox/mesh-sdk/go/composer"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
)
|
||||
|
||||
// ManifestsState is the delivery indexer's manifests by digest, which a compose ask's request names (the module reads
|
||||
// it, module.json reads).
|
||||
const ManifestsState = delivery.IndexerModule + ".manifests"
|
||||
|
||||
// composerLocks hold one mutex per composer key, so a composer two asks need at once is built once.
|
||||
var composerLocks sync.Map
|
||||
|
||||
// composerPath is where the composer of key is kept: the cache directory, beside Go's own caches.
|
||||
func (m *Module) composerPath(key string) string {
|
||||
return filepath.Join(m.Cache, "composers", key, "mesh-compose")
|
||||
}
|
||||
|
||||
// compose runs a compose ask: the composer of its commit, built once, over its request with the manifests filled.
|
||||
func (m *Module) compose(ctx context.Context, fetch buildstep.StepFunc, a delivery.StepAsk) (any, error) {
|
||||
job := *a.Compose
|
||||
repository, _ := job.Fetch["repository"].(string)
|
||||
commit, _ := job.Fetch["commit"].(string)
|
||||
installed, err := m.Probe(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
key := delivery.ComposerKey(repository, commit, job.Dir, job.Package, delivery.ToolchainUsed{Version: installed, Recipe: m.Recipe.Version()})
|
||||
exe, built, err := m.composerOf(ctx, fetch, job, key)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
req, err := composer.Decode(job.Request)
|
||||
if err != nil {
|
||||
return nil, buildstep.Refuse("the compose ask %s: %v", a.ID, err)
|
||||
}
|
||||
if err := composer.Fill(&req, m.manifestOf); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
verdict, err := m.runComposer(ctx, exe, req)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return delivery.ComposeAnswer{Built: built, Key: key, Verdict: verdict}, nil
|
||||
}
|
||||
|
||||
// manifestOf is the manifest the indexer keeps under digest, as its repository holds it.
|
||||
func (m *Module) manifestOf(digest string) ([]byte, error) {
|
||||
if m.ReadManifest == nil {
|
||||
return nil, errors.New("this module reads no manifests state")
|
||||
}
|
||||
key, err := delivery.ManifestsKey(digest)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
raw, err := m.ReadManifest(key)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if raw == nil {
|
||||
return nil, fmt.Errorf("the indexer keeps no manifest of digest %s", digest)
|
||||
}
|
||||
man, err := delivery.DecodeManifest(raw)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return []byte(man.JSON), nil
|
||||
}
|
||||
|
||||
// composerOf is the composer kept under key, or, when none is, the one built now from the fetched commit (built true).
|
||||
func (m *Module) composerOf(ctx context.Context, fetch buildstep.StepFunc, job delivery.ComposeJob, key string) (string, bool, error) {
|
||||
exe := m.composerPath(key)
|
||||
mu, _ := composerLocks.LoadOrStore(exe, &sync.Mutex{})
|
||||
mu.(*sync.Mutex).Lock()
|
||||
defer mu.(*sync.Mutex).Unlock()
|
||||
if st, err := os.Stat(exe); err == nil && st.Mode().IsRegular() {
|
||||
return exe, false, nil
|
||||
}
|
||||
if fetch == nil {
|
||||
return "", false, buildstep.Refuse("this build module fetches nothing")
|
||||
}
|
||||
out, err := fetch(ctx, job.Fetch)
|
||||
if err != nil {
|
||||
return "", false, err
|
||||
}
|
||||
raw, err := json.Marshal(out)
|
||||
if err != nil {
|
||||
return "", false, err
|
||||
}
|
||||
var f struct {
|
||||
Workspace string `json:"workspace"`
|
||||
}
|
||||
if err := json.Unmarshal(raw, &f); err != nil || f.Workspace == "" {
|
||||
return "", false, fmt.Errorf("the fetch answered no workspace: %s", raw)
|
||||
}
|
||||
defer m.releaseSaid(f.Workspace)
|
||||
done := m.spaces().Hold(f.Workspace)
|
||||
defer done()
|
||||
dir := filepath.Join(m.Work, f.Workspace, "tree", job.Dir)
|
||||
if _, err := os.Stat(filepath.Join(dir, "go.mod")); err != nil {
|
||||
return "", false, fmt.Errorf("%s at %s holds no go.mod in %s, so no composer can be built there", job.Fetch["repository"], job.Fetch["commit"], job.Dir)
|
||||
}
|
||||
into := filepath.Join(m.Work, f.Workspace, "composer", "mesh-compose")
|
||||
if err := os.MkdirAll(filepath.Dir(into), 0o700); err != nil {
|
||||
return "", false, err
|
||||
}
|
||||
run, err := m.stepRunner(f.Workspace, false)
|
||||
if err != nil {
|
||||
return "", false, err
|
||||
}
|
||||
env := append(m.goEnvFor(dir), "GOPROXY=off", "GOTOOLCHAIN=local", "CGO_ENABLED=0")
|
||||
r := run(ctx, env, "go", "-C", dir, "build", "-trimpath", "-o", into, job.Package)
|
||||
if r.Status != 0 || r.Err != "" {
|
||||
// The words Go says when the module holds no such package (a mesh-sdk from before the composer) are kept whole,
|
||||
// so the asker tells that apart from a fault.
|
||||
return "", false, fmt.Errorf("the composer %s does not build at %s: %s", job.Package, job.Fetch["commit"], compileError(r.Err+"\n"+r.Stderr+"\n"+r.Stdout))
|
||||
}
|
||||
if err := copyExecutable(into, exe); err != nil {
|
||||
return "", false, fmt.Errorf("the composer could not be kept: %w", err)
|
||||
}
|
||||
return exe, true, nil
|
||||
}
|
||||
|
||||
// copyExecutable copies a built program into place through a file beside it, so a half-written program is never kept.
|
||||
func copyExecutable(from, to string) error {
|
||||
if err := os.MkdirAll(filepath.Dir(to), 0o700); err != nil {
|
||||
return err
|
||||
}
|
||||
src, err := os.Open(from)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer src.Close()
|
||||
tmp := to + ".new"
|
||||
dst, err := os.OpenFile(tmp, os.O_CREATE|os.O_TRUNC|os.O_WRONLY, 0o700)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if _, err := io.Copy(dst, src); err != nil {
|
||||
dst.Close()
|
||||
return err
|
||||
}
|
||||
if err := dst.Close(); err != nil {
|
||||
return err
|
||||
}
|
||||
return os.Rename(tmp, to)
|
||||
}
|
||||
|
||||
// releaseSaid removes a workspace once its steps ended, saying a failure: the sweep removes it later.
|
||||
func (m *Module) releaseSaid(workspace string) {
|
||||
if err := m.spaces().Release(workspace); err != nil && m.Say != nil {
|
||||
m.Say("the compose ask's workspace %s: %v", workspace, err)
|
||||
}
|
||||
}
|
||||
|
||||
// runComposer runs the composer over req in a workspace of its own, shut off, and answers its verdict: the last line of
|
||||
// its standard output, a composer.Verdict of the version this module reads.
|
||||
func (m *Module) runComposer(ctx context.Context, exe string, req composer.Request) (json.RawMessage, error) {
|
||||
id := buildstep.NewID()
|
||||
done := m.spaces().Hold(id)
|
||||
defer m.releaseSaid(id)
|
||||
defer done()
|
||||
ws := filepath.Join(m.Work, id)
|
||||
if err := os.MkdirAll(ws, 0o700); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
raw, err := json.Marshal(req)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
file := filepath.Join(ws, "request.json")
|
||||
if err := os.WriteFile(file, raw, 0o600); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
run := m.Run
|
||||
if !m.Inside {
|
||||
iso := m.isolation(id, false)
|
||||
iso.ReadOnly = append(iso.ReadOnly, exe)
|
||||
run = buildstep.Isolated(m.Run, iso)
|
||||
}
|
||||
r := run(ctx, nil, exe, "-request", file)
|
||||
line := strings.TrimSpace(lastLine(r.Stdout))
|
||||
var v composer.Verdict
|
||||
if r.Status != 0 || r.Err != "" || json.Unmarshal([]byte(line), &v) != nil || v.V != composer.Version {
|
||||
return nil, fmt.Errorf("the composer answered no verdict (status %d %s): %s", r.Status, r.Err, firstLine(r.Stderr+"\n"+r.Stdout))
|
||||
}
|
||||
return json.RawMessage(line), nil
|
||||
}
|
||||
@@ -0,0 +1,156 @@
|
||||
//go:build integration
|
||||
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"os"
|
||||
"os/exec"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/nats-io/nats.go"
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/buildstep"
|
||||
"git.novox.be/novox/mesh-sdk/go/composer"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"git.novox.be/novox/mesh-sdk/go/forgekey"
|
||||
"git.novox.be/novox/mesh-sdk/go/testbus"
|
||||
)
|
||||
|
||||
// sdkPin is the mesh-sdk version this module's go.mod requires: the composer the test builds is that version's.
|
||||
func sdkPin(t *testing.T) string {
|
||||
t.Helper()
|
||||
raw, err := os.ReadFile("../../go.mod")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for _, l := range strings.Split(string(raw), "\n") {
|
||||
if f := strings.Fields(l); len(f) >= 2 && f[0] == "git.novox.be/novox/mesh-sdk/go" {
|
||||
return f[1]
|
||||
}
|
||||
}
|
||||
t.Fatal("go.mod requires no mesh-sdk")
|
||||
return ""
|
||||
}
|
||||
|
||||
// goEnvOf is one value of this machine's go env.
|
||||
func goEnvOf(t *testing.T, name string) string {
|
||||
t.Helper()
|
||||
out, err := exec.Command("go", "env", name).Output()
|
||||
if err != nil {
|
||||
t.Fatalf("go env %s: %v", name, err)
|
||||
}
|
||||
return strings.TrimSpace(string(out))
|
||||
}
|
||||
|
||||
// 1622/AC1, AC3 against the real bus: a compose ask put on the Go build seat's check stream is pulled by the module
|
||||
// started whole on that bus, which fetches the commit (here a tree of a module requiring mesh-sdk at this module's pin,
|
||||
// as a reader repository is), builds mesh-sdk's composer there with nothing fetched, fills the request's manifests from
|
||||
// the indexer's state and runs the composer; the step key <node>.<ask id> passes with the composer's verdict. A second
|
||||
// ask of the same commit runs the kept composer: fetched and built once.
|
||||
func TestAComposeAskIsBuiltOnceAndAnsweredOnTheBus(t *testing.T) {
|
||||
nc, err := nats.Connect(testbus.URL(t))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer nc.Close()
|
||||
js, err := jetstream.New(nc)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
ctx, stop := context.WithCancel(context.Background())
|
||||
defer stop()
|
||||
if err := buildstep.DeclareAskWorkers(ctx, js, delivery.AskGo, 2*time.Minute); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() { _ = js.DeleteStream(context.Background(), "SEAT_NODE_BUILD_GO") })
|
||||
|
||||
states := newStates()
|
||||
m := moduleWith(states, "go1.26.0")
|
||||
m.Work, m.Cache, m.Inside = t.TempDir(), t.TempDir(), true
|
||||
m.Run = StepRunner(t.TempDir())
|
||||
m.GoModCache = goEnvOf(t, "GOMODCACHE") // the module cache this module's own build filled, read and not fetched
|
||||
run, gocache := m.Run, goEnvOf(t, "GOCACHE") // this machine's build cache, so the composer's packages compile once
|
||||
m.Run = func(ctx context.Context, env []string, name string, args ...string) forgekey.Ran {
|
||||
return run(ctx, append(env, "GOCACHE="+gocache), name, args...)
|
||||
}
|
||||
read, digests := indexerHolds(t, `{"module":"demo","version":"1"}`)
|
||||
m.ReadManifest = read
|
||||
|
||||
gosum, err := os.ReadFile("../../go.sum")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
fetches := 0
|
||||
steps := fakeSteps(new([]string))
|
||||
for i := range steps {
|
||||
if steps[i].Name == "build_go_fetch" {
|
||||
steps[i].Run = func(ctx context.Context, args map[string]any) (any, error) {
|
||||
fetches++
|
||||
ws := buildstep.NewID()
|
||||
tree := filepath.Join(m.Work, ws, "tree")
|
||||
if err := os.MkdirAll(tree, 0o700); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
mod := "module reader\n\ngo 1.26.0\n\nrequire git.novox.be/novox/mesh-sdk/go " + sdkPin(t) + "\n"
|
||||
if err := os.WriteFile(filepath.Join(tree, "go.mod"), []byte(mod), 0o600); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return map[string]any{"workspace": ws}, os.WriteFile(filepath.Join(tree, "go.sum"), gosum, 0o600)
|
||||
}
|
||||
}
|
||||
}
|
||||
p := m.Puller(steps, &buildstep.JetStreamAsks{JS: js, Language: delivery.AskGo})
|
||||
p.PullWait = 200 * time.Millisecond
|
||||
go p.Pull(ctx)
|
||||
|
||||
ask := func() delivery.Step {
|
||||
a := composeAskOf(t, digests[0])
|
||||
a.Compose.Fetch["repository"], a.Compose.Dir = "novox/mesh-controller", delivery.RootPath
|
||||
raw, err := delivery.EncodeStepAsk(a)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
subject, _ := delivery.AskSubject(delivery.AskGo, delivery.AskCheck)
|
||||
if _, err := js.Publish(ctx, subject, raw); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
key, _ := delivery.AskStepKey("ace", a.ID)
|
||||
for deadline := time.Now().Add(90 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
|
||||
states.mu.Lock()
|
||||
v := states.steps[key]
|
||||
states.mu.Unlock()
|
||||
if len(v) > 0 && v[len(v)-1].State != delivery.StepRunning {
|
||||
return v[len(v)-1]
|
||||
}
|
||||
}
|
||||
t.Fatalf("the compose ask %s never ended", a.ID)
|
||||
return delivery.Step{}
|
||||
}
|
||||
for n, wantBuilt := range []bool{true, false} {
|
||||
step := ask()
|
||||
if step.State != delivery.StepPassed {
|
||||
t.Fatalf("ask %d: %s: %s", n+1, step.State, step.Reason)
|
||||
}
|
||||
var ans delivery.ComposeAnswer
|
||||
var v composer.Verdict
|
||||
if err := json.Unmarshal(step.Result, &ans); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := json.Unmarshal(ans.Verdict, &v); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if ans.Built != wantBuilt || !strings.HasPrefix(ans.Key, "composer-") || len(v.Cases) != 1 || !v.Cases[0].Manifests["demo"].Accepted ||
|
||||
!strings.Contains(v.SDK, sdkPin(t)) {
|
||||
t.Fatalf("ask %d: answer built %v key %s, verdict %+v; want built %v and demo accepted by mesh-sdk %s", n+1, ans.Built, ans.Key, v, wantBuilt, sdkPin(t))
|
||||
}
|
||||
}
|
||||
if fetches != 1 {
|
||||
t.Errorf("two asks of one commit fetched %d times, want once", fetches)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,125 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"slices"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/buildstep"
|
||||
"git.novox.be/novox/mesh-sdk/go/composer"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
)
|
||||
|
||||
const composeCommit = "4444444444444444444444444444444444444444"
|
||||
|
||||
// indexerHolds is the delivery indexer's manifests state holding each body under its digest, as the indexer writes it.
|
||||
func indexerHolds(t *testing.T, bodies ...string) (func(string) (json.RawMessage, error), []string) {
|
||||
t.Helper()
|
||||
kept, digests := map[string]json.RawMessage{}, []string{}
|
||||
for _, b := range bodies {
|
||||
d := delivery.ManifestDigest([]byte(b))
|
||||
key, _ := delivery.ManifestsKey(d)
|
||||
raw, err := delivery.EncodeManifest(delivery.Manifest{Module: "demo", JSON: b})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
kept[key], digests = raw, append(digests, d)
|
||||
}
|
||||
return func(key string) (json.RawMessage, error) { return kept[key], nil }, digests
|
||||
}
|
||||
|
||||
func composeAskOf(t *testing.T, digest string) delivery.StepAsk {
|
||||
t.Helper()
|
||||
req, _ := json.Marshal(composer.Request{V: composer.Version, Cases: []composer.Case{{Name: "change",
|
||||
Manifests: map[string]composer.Manifest{"demo": {Digest: digest}}}}})
|
||||
return delivery.StepAsk{ID: buildstep.NewID(), Kind: delivery.AskCheck, Language: delivery.AskGo, Asker: "mesh-delivery-assigner",
|
||||
At: time.Now(), Compose: &delivery.ComposeJob{Fetch: map[string]any{"repository": "novox/mesh-sdk", "commit": composeCommit},
|
||||
Dir: "go", Package: composer.Package, Request: req}}
|
||||
}
|
||||
|
||||
// composeModule is a module with work and cache directories of its own, running its commands as they are (Inside: a
|
||||
// unit test has no bubblewrap to shut them off in).
|
||||
func composeModule(t *testing.T) *Module {
|
||||
t.Helper()
|
||||
m := moduleWith(newStates(), "go1.26.0")
|
||||
m.Work, m.Cache, m.Inside, m.Run = t.TempDir(), t.TempDir(), true, StepRunner(t.TempDir())
|
||||
return m
|
||||
}
|
||||
|
||||
// 1622/AC3: a composer kept under the ask's key is run again, nothing fetched or built: the answer says it was not
|
||||
// built, names the key, and carries the composer's verdict whole; the request it read had its manifests filled from the
|
||||
// indexer's state.
|
||||
func TestAKeptComposerIsRunWithTheManifestsFilled(t *testing.T) {
|
||||
m := composeModule(t)
|
||||
read, digests := indexerHolds(t, `{"module":"demo","version":"1"}`)
|
||||
m.ReadManifest = read
|
||||
a := composeAskOf(t, digests[0])
|
||||
key := delivery.ComposerKey("novox/mesh-sdk", composeCommit, "go", composer.Package, delivery.ToolchainUsed{Version: "go1.26.0", Recipe: m.Recipe.Version()})
|
||||
exe := m.composerPath(key)
|
||||
if err := os.MkdirAll(filepath.Dir(exe), 0o700); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
// A composer that answers which manifests its request held, filled or not, as a verdict.
|
||||
script := "#!/bin/sh\n" +
|
||||
`if grep -q '"digest"' "$2"; then echo '{"v":1,"sdk":"stand-in","cases":[{"name":"unfilled","manifests":{}}]}'; ` +
|
||||
`else echo '{"v":1,"sdk":"stand-in","cases":[{"name":"change","manifests":{"demo":{"accepted":true}}}]}'; fi` + "\n"
|
||||
if err := os.WriteFile(exe, []byte(script), 0o700); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
fetched := 0
|
||||
fetch := func(context.Context, map[string]any) (any, error) { fetched++; return nil, nil }
|
||||
out, err := m.compose(context.Background(), fetch, a)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
ans := out.(delivery.ComposeAnswer)
|
||||
var v composer.Verdict
|
||||
if err := json.Unmarshal(ans.Verdict, &v); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if ans.Built || ans.Key != key || fetched != 0 || len(v.Cases) != 1 || v.Cases[0].Name != "change" || !v.Cases[0].Manifests["demo"].Accepted {
|
||||
t.Fatalf("answer = %+v (fetched %d), verdict %+v", ans, fetched, v)
|
||||
}
|
||||
if entries, _ := os.ReadDir(m.Work); len(entries) != 0 {
|
||||
t.Errorf("the run's workspace was left: %v", entries)
|
||||
}
|
||||
}
|
||||
|
||||
// 1622: a manifest the indexer does not keep fails the ask, naming its digest; a fetched commit with no go.mod where
|
||||
// the ask builds fails it naming the directory, and its workspace is released.
|
||||
func TestAComposeAskFailsLoudly(t *testing.T) {
|
||||
m := composeModule(t)
|
||||
m.ReadManifest, _ = indexerHolds(t)
|
||||
a := composeAskOf(t, "feedfeedfeedfeedfeedfeedfeedfeedfeedfeedfeedfeedfeedfeedfeedfeed")
|
||||
key := delivery.ComposerKey("novox/mesh-sdk", composeCommit, "go", composer.Package, delivery.ToolchainUsed{Version: "go1.26.0", Recipe: m.Recipe.Version()})
|
||||
os.MkdirAll(filepath.Dir(m.composerPath(key)), 0o700)
|
||||
os.WriteFile(m.composerPath(key), []byte("#!/bin/sh\nexit 0\n"), 0o700)
|
||||
if _, err := m.compose(context.Background(), nil, a); err == nil || !strings.Contains(err.Error(), "feedfeed") {
|
||||
t.Errorf("an unkept manifest: %v", err)
|
||||
}
|
||||
m2 := composeModule(t)
|
||||
fetch := func(context.Context, map[string]any) (any, error) {
|
||||
if err := os.MkdirAll(filepath.Join(m2.Work, "ws0000000000a", "tree", "go"), 0o700); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return map[string]any{"workspace": "ws0000000000a"}, nil
|
||||
}
|
||||
if _, err := m2.compose(context.Background(), fetch, composeAskOf(t, "x")); err == nil || !strings.Contains(err.Error(), "no go.mod in go") {
|
||||
t.Errorf("a commit with no go.mod: %v", err)
|
||||
}
|
||||
if entries, _ := os.ReadDir(m2.Work); len(entries) != 0 {
|
||||
t.Errorf("the fetched workspace was left: %v", entries)
|
||||
}
|
||||
}
|
||||
|
||||
// 1622: the module reads the delivery indexer's manifests, which a compose ask names by digest.
|
||||
func TestTheModuleReadsTheIndexersManifests(t *testing.T) {
|
||||
if m := readManifest(t); !slices.Contains(m.Reads, ManifestsState) {
|
||||
t.Errorf("reads = %v, lacks %s", m.Reads, ManifestsState)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,112 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"os"
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// A failed test keeps the lines that say why it failed, not the lines it logged first (issue 1901). The fixture is
|
||||
// the real `go test -json` of a test that logs a line and then fails with a multi-line t.Fatalf.
|
||||
func TestAFailedTestKeepsItsFatalMessageAfterItsLogLines(t *testing.T) {
|
||||
raw, err := os.ReadFile("testdata/captured/go-test-json-log-then-fatal.jsonl")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
res := &TestResult{}
|
||||
rd := newReader(res, ".")
|
||||
rd.read(string(raw))
|
||||
rd.finish()
|
||||
if len(res.Failed) != 1 {
|
||||
t.Fatalf("failures = %+v, want one", res.Failed)
|
||||
}
|
||||
f := res.Failed[0]
|
||||
for _, want := range []string{"3 subscription(s) derive to a subject no emitter owns:", "a.b.c", "d.e.f", "g.h.i"} {
|
||||
if !strings.Contains(f.Output, want) {
|
||||
t.Errorf("1901/AC1: output lacks %q: %q", want, f.Output)
|
||||
}
|
||||
}
|
||||
if want := "3 subscription(s) derive to a subject no emitter owns:"; f.Cause != want && !strings.HasSuffix(f.Cause, want) {
|
||||
t.Errorf("1901/AC1: cause = %q, want the t.Fatalf line, not the t.Logf line", f.Cause)
|
||||
}
|
||||
}
|
||||
|
||||
// A failure longer than the bound keeps its last lines, where the cause is, and says how many earlier lines it dropped.
|
||||
func TestAFailureLongerThanTheBoundKeepsItsLastLines(t *testing.T) {
|
||||
var lines []string
|
||||
for i := 0; i < 400; i++ {
|
||||
lines = append(lines, " x_test.go:1: log line "+strings.Repeat("y", 40))
|
||||
}
|
||||
lines = append(lines, " x_test.go:9: the real cause", "--- FAIL: TestX (0.00s)")
|
||||
out := clip(lines, "")
|
||||
if len(out) > outputBytes+100 {
|
||||
t.Errorf("1901/AC1: kept %d bytes, bound is %d", len(out), outputBytes)
|
||||
}
|
||||
if !strings.HasSuffix(out, "the real cause\n--- FAIL: TestX (0.00s)") || !strings.Contains(out, "\n... (") {
|
||||
t.Errorf("1901/AC1: out = %q...%q", out[:40], out[len(out)-60:])
|
||||
}
|
||||
}
|
||||
|
||||
// An older checker, which knows no "cause" field, still reads the new build module's answer, and the output it gets
|
||||
// holds at least what the older build module kept (its first lines, for an output shorter than the bound).
|
||||
func TestAnOlderCheckerReadsTheNewFailureAndGetsAtLeastWhatItKept(t *testing.T) { // 1901/AC3
|
||||
raw, err := os.ReadFile("testdata/captured/go-test-json-log-then-fatal.jsonl")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
res := &TestResult{}
|
||||
rd := newReader(res, ".")
|
||||
rd.read(string(raw))
|
||||
rd.finish()
|
||||
b, err := json.Marshal(res.Failed)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var old []struct {
|
||||
Package string `json:"package"`
|
||||
Test string `json:"test,omitempty"`
|
||||
Output string `json:"output,omitempty"`
|
||||
}
|
||||
if err := json.Unmarshal(b, &old); err != nil || len(old) != 1 {
|
||||
t.Fatalf("1901/AC3: an older checker refused the answer: %v, %d failures", err, len(old))
|
||||
}
|
||||
first := strings.SplitN(old[0].Output, "\n", 2)[0]
|
||||
if first == "" || !strings.Contains(old[0].Output, "judged against") || !strings.Contains(old[0].Output, "3 subscription(s)") {
|
||||
t.Errorf("1901/AC3: older checker's output = %q", old[0].Output)
|
||||
}
|
||||
}
|
||||
|
||||
// A long failure, whose earlier lines are left out, still starts with a real line for an older checker, which keeps only the
|
||||
// first non-empty line of the output cut to 120 characters, and not with the marker.
|
||||
func TestALongFailureStillStartsWithARealLineForAnOlderChecker(t *testing.T) { // 1901/AC3
|
||||
var lines []string
|
||||
for i := 0; i < 400; i++ {
|
||||
lines = append(lines, " x_test.go:1: log line "+strings.Repeat("y", 40))
|
||||
}
|
||||
lines = append(lines, " x_test.go:9: the real cause", "--- FAIL: TestX (0.00s)")
|
||||
oldFirst := func(out string) string { // the older checker's rule (judge.go firstLine)
|
||||
for _, l := range strings.Split(out, "\n") {
|
||||
if l = strings.TrimSpace(l); l != "" {
|
||||
if len(l) > 120 {
|
||||
l = l[:120]
|
||||
}
|
||||
return l
|
||||
}
|
||||
}
|
||||
return ""
|
||||
}
|
||||
for _, cause := range []string{"the real cause", ""} {
|
||||
out := clip(lines, cause)
|
||||
got := oldFirst(out)
|
||||
if strings.HasPrefix(got, "... (") || got == "" {
|
||||
t.Errorf("1901/AC3: cause %q: an older checker would read %q", cause, got)
|
||||
}
|
||||
if cause != "" && got != cause {
|
||||
t.Errorf("1901/AC3: older checker reads %q, want the cause", got)
|
||||
}
|
||||
if !strings.HasSuffix(out, "--- FAIL: TestX (0.00s)") {
|
||||
t.Errorf("1901/AC3: the last lines are lost: %q", out[len(out)-40:])
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -288,10 +288,14 @@ func inTestdata(path string) bool {
|
||||
// git, through the account's rewrite of the forge's address to ssh), and nothing that asks a question. For a directory
|
||||
// that vendors, see goEnvFor.
|
||||
func (m *Module) goEnv() []string {
|
||||
modcache := filepath.Join(m.Cache, "gomod")
|
||||
if m.GoModCache != "" {
|
||||
modcache = m.GoModCache
|
||||
}
|
||||
return []string{
|
||||
"GOTOOLCHAIN=local",
|
||||
"GOFLAGS=-mod=mod",
|
||||
"GOMODCACHE=" + filepath.Join(m.Cache, "gomod"),
|
||||
"GOMODCACHE=" + modcache,
|
||||
"GOCACHE=" + filepath.Join(m.Cache, "gobuild"),
|
||||
"GOPRIVATE=git.novox.be",
|
||||
"GIT_TERMINAL_PROMPT=0",
|
||||
|
||||
@@ -9,6 +9,7 @@ import (
|
||||
"bufio"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"regexp"
|
||||
"sort"
|
||||
"strings"
|
||||
"time"
|
||||
@@ -49,11 +50,13 @@ type TestTime struct {
|
||||
}
|
||||
|
||||
// Failure is a test that failed, or a package that failed with no test to name (a build that failed, a TestMain that
|
||||
// panicked); Output is the first lines it wrote.
|
||||
// panicked); Output is the last lines it wrote, up to outputBytes (issue 1901), and Cause the first line of what the
|
||||
// test reported as its error (t.Error, t.Fatal), without its file and line, when go test said which lines those were.
|
||||
type Failure struct {
|
||||
Package string `json:"package"`
|
||||
Test string `json:"test,omitempty"`
|
||||
Output string `json:"output,omitempty"`
|
||||
Cause string `json:"cause,omitempty"`
|
||||
}
|
||||
|
||||
// TestResult is what a test step passes with.
|
||||
@@ -96,13 +99,20 @@ type event struct {
|
||||
Package string
|
||||
Test string
|
||||
Output string
|
||||
OutputType string // "error" for a t.Error or t.Fatal message, "error-continue" for its further lines (go test -json)
|
||||
ImportPath string
|
||||
FailedBuild string
|
||||
Elapsed float64
|
||||
}
|
||||
|
||||
// outputLines is how many lines of output a failure keeps.
|
||||
const outputLines = 15
|
||||
// outputBytes is the most a failure keeps of its output, and outputLineBytes the most of one line. **A failure keeps its
|
||||
// last lines** (issue 1901): a test logs before it fails, so the lines that say why are at the end, and the first
|
||||
// fifteen lines kept before lost them (the verdict of group r4g kept a t.Logf line, not the t.Fatalf after it). Four
|
||||
// kilobytes holds a long assertion message and stays far under a bus message's or a verdict's bound.
|
||||
const (
|
||||
outputBytes = 4096
|
||||
outputLineBytes = 1024
|
||||
)
|
||||
|
||||
// reader collects the events of one go.mod directory's `go test -json` into a result.
|
||||
type reader struct {
|
||||
@@ -113,13 +123,14 @@ type reader struct {
|
||||
builds map[string][]string // build output by import path
|
||||
pkgOut map[string][]string
|
||||
testOut map[string][]string // by package + "\x00" + test
|
||||
causes map[string]string // the first error line of a test, by the same key
|
||||
times []TestTime
|
||||
failedIn map[string]bool // packages with a failed test
|
||||
}
|
||||
|
||||
func newReader(res *TestResult, dir string) *reader {
|
||||
return &reader{res: res, dir: dir, pkgs: map[string]*PackageResult{}, builds: map[string][]string{},
|
||||
pkgOut: map[string][]string{}, testOut: map[string][]string{}, failedIn: map[string]bool{}}
|
||||
pkgOut: map[string][]string{}, testOut: map[string][]string{}, causes: map[string]string{}, failedIn: map[string]bool{}}
|
||||
}
|
||||
|
||||
// read takes go test's output. Lines that are not events are kept for a failure that has nothing else to say.
|
||||
@@ -183,6 +194,9 @@ func (r *reader) event(ev event) {
|
||||
switch ev.Action {
|
||||
case "output":
|
||||
r.testOut[key] = keep(r.testOut[key], ev.Output)
|
||||
if ev.OutputType == "error" && r.causes[key] == "" {
|
||||
r.causes[key] = errorWords(ev.Output)
|
||||
}
|
||||
case "pass", "fail", "skip":
|
||||
p := r.pkg(ev.Package)
|
||||
p.Tests++
|
||||
@@ -193,17 +207,56 @@ func (r *reader) event(ev event) {
|
||||
r.times = append(r.times, TestTime{Package: ev.Package, Test: ev.Test, Elapsed: ev.Elapsed})
|
||||
if ev.Action == "fail" {
|
||||
r.failedIn[ev.Package] = true
|
||||
r.res.Failed = append(r.res.Failed, Failure{Package: ev.Package, Test: ev.Test, Output: clip(r.testOut[key])})
|
||||
r.res.Failed = append(r.res.Failed, Failure{Package: ev.Package, Test: ev.Test, Output: clip(r.testOut[key], r.causes[key]), Cause: r.causes[key]})
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func clip(lines []string) string {
|
||||
if len(lines) > outputLines {
|
||||
lines = append(lines[:outputLines:outputLines], fmt.Sprintf("... (%d more lines)", len(lines)-outputLines))
|
||||
// clip joins the last lines of a failure's output that fit outputBytes, each cut to outputLineBytes, and says how many
|
||||
// earlier lines it left out. **When it leaves lines out, its first line is still a real one** (issue 1901/AC3): the failure's
|
||||
// cause, or failing that the first non-empty line of the output, then the marker. An older checker takes only the first
|
||||
// non-empty line of the output, and a marker there would be less than the older pair kept.
|
||||
func clip(lines []string, cause string) string {
|
||||
size, from := 0, len(lines)
|
||||
for from > 0 {
|
||||
n := len(cutLine(lines[from-1])) + 1
|
||||
if size+n > outputBytes && from < len(lines) {
|
||||
break
|
||||
}
|
||||
size += n
|
||||
from--
|
||||
}
|
||||
return strings.Join(lines, "\n")
|
||||
kept := make([]string, 0, len(lines)-from+1)
|
||||
if from > 0 {
|
||||
head := strings.TrimSpace(cause)
|
||||
for i := 0; head == "" && i < from; i++ {
|
||||
head = strings.TrimSpace(lines[i])
|
||||
}
|
||||
if head != "" {
|
||||
kept = append(kept, cutLine(head))
|
||||
}
|
||||
kept = append(kept, fmt.Sprintf("... (%d earlier lines)", from))
|
||||
}
|
||||
for _, l := range lines[from:] {
|
||||
kept = append(kept, cutLine(l))
|
||||
}
|
||||
return strings.Join(kept, "\n")
|
||||
}
|
||||
|
||||
// cutLine cuts a line to outputLineBytes, on a character boundary.
|
||||
func cutLine(l string) string {
|
||||
if len(l) <= outputLineBytes {
|
||||
return l
|
||||
}
|
||||
return strings.ToValidUTF8(l[:outputLineBytes], "") + "..."
|
||||
}
|
||||
|
||||
var fileAndLine = regexp.MustCompile(`^\S+\.go:\d+: `)
|
||||
|
||||
// errorWords is the words of an error line of a test: trimmed, without the file and line go test put before them.
|
||||
func errorWords(out string) string {
|
||||
return strings.TrimSpace(fileAndLine.ReplaceAllString(strings.TrimSpace(out), ""))
|
||||
}
|
||||
|
||||
// finish moves what was read into the result: the packages in order, a failure for each failed package that has no
|
||||
@@ -212,7 +265,7 @@ func (r *reader) finish() {
|
||||
for _, name := range r.order {
|
||||
p := r.pkgs[name]
|
||||
if !p.Passed && !r.failedIn[name] {
|
||||
r.res.Failed = append(r.res.Failed, Failure{Package: name, Output: clip(r.pkgOut[name])})
|
||||
r.res.Failed = append(r.res.Failed, Failure{Package: name, Output: clip(r.pkgOut[name], "")})
|
||||
}
|
||||
if !p.NoTests && p.Elapsed > r.res.Elapsed {
|
||||
r.res.Elapsed = p.Elapsed
|
||||
|
||||
@@ -12,6 +12,7 @@ import (
|
||||
"regexp"
|
||||
"sort"
|
||||
"strings"
|
||||
"sync"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/forgekey"
|
||||
)
|
||||
@@ -127,3 +128,45 @@ func (m *Module) lint(ctx context.Context, run forgekey.Runner, tree, d string,
|
||||
}
|
||||
return failed, nil
|
||||
}
|
||||
|
||||
// lintAll lints each directory, up to m.parallel() at once like the level's tests; the failures join in directory order
|
||||
// and the error is the first directory's, so the answer does not depend on which ended first (issue 1930).
|
||||
func (m *Module) lintAll(ctx context.Context, run forgekey.Runner, tree string, dirs []string, env func(d string) []string) ([]Failure, error) {
|
||||
type answer struct {
|
||||
failed []Failure
|
||||
err error
|
||||
}
|
||||
answers := make([]answer, len(dirs))
|
||||
slots := make(chan struct{}, m.parallel())
|
||||
var wg sync.WaitGroup
|
||||
var crash any // a panic of a lint goes on in the caller, as a panic of a test run does
|
||||
var crashMu sync.Mutex
|
||||
for i, d := range dirs {
|
||||
wg.Add(1)
|
||||
slots <- struct{}{}
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
defer func() { <-slots }()
|
||||
defer func() {
|
||||
if p := recover(); p != nil {
|
||||
crashMu.Lock()
|
||||
crash = p
|
||||
crashMu.Unlock()
|
||||
}
|
||||
}()
|
||||
answers[i].failed, answers[i].err = m.lint(ctx, run, tree, d, env(d))
|
||||
}()
|
||||
}
|
||||
wg.Wait()
|
||||
if crash != nil {
|
||||
panic(crash)
|
||||
}
|
||||
var failed []Failure
|
||||
for _, a := range answers {
|
||||
if a.err != nil {
|
||||
return nil, a.err
|
||||
}
|
||||
failed = append(failed, a.failed...)
|
||||
}
|
||||
return failed, nil
|
||||
}
|
||||
|
||||
@@ -5,7 +5,9 @@ import (
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/forgekey"
|
||||
)
|
||||
@@ -90,3 +92,43 @@ func TestAnUnformattedFileFailsTheUnitLevel(t *testing.T) {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Issue 1930: the lint of many directories runs side by side up to the level's bound, not one after another, and the
|
||||
// failures still join in directory order.
|
||||
func TestTheLintOfManyDirectoriesRunsAtOnceUpToTheBound(t *testing.T) {
|
||||
files := map[string]string{}
|
||||
var dirs []string
|
||||
for _, d := range []string{"a", "b", "c", "d", "e", "f"} {
|
||||
files[d+"/go.mod"] = "module x/" + d + "\n"
|
||||
files[d+"/x.go"] = "package x\n"
|
||||
dirs = append(dirs, d)
|
||||
}
|
||||
tree := writeTree(t, files)
|
||||
var mu sync.Mutex
|
||||
running, most := 0, 0
|
||||
run := func(ctx context.Context, env []string, name string, args ...string) forgekey.Ran {
|
||||
mu.Lock()
|
||||
running++
|
||||
most = max(most, running)
|
||||
mu.Unlock()
|
||||
time.Sleep(30 * time.Millisecond)
|
||||
mu.Lock()
|
||||
running--
|
||||
mu.Unlock()
|
||||
if strings.HasSuffix(args[1], "b") || strings.HasSuffix(args[1], "e") {
|
||||
return forgekey.Ran{Status: 1, Stderr: "bad " + filepath.Base(args[1])}
|
||||
}
|
||||
return forgekey.Ran{}
|
||||
}
|
||||
m := &Module{Parallel: 3}
|
||||
failed, err := m.lintAll(context.Background(), run, tree, dirs, func(string) []string { return nil })
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if most < 2 || most > 3 {
|
||||
t.Errorf("%d lint commands ran at once, want 2 or 3 (the bound)", most)
|
||||
}
|
||||
if len(failed) != 2 || failed[0].Package != "b" || failed[1].Package != "e" {
|
||||
t.Errorf("failed = %+v, want b's then e's", failed)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -54,6 +54,7 @@ func main() {
|
||||
m.RegistryFile = m.Home + "/registry"
|
||||
m.Settings = m.Home + "/settings.json"
|
||||
m.ReadKept = readOf(stdio.State(KeptState)).Get
|
||||
m.ReadManifest = readOf(stdio.State(ManifestsState)).Get
|
||||
m.Probe = ProbeGo(m.Home, m.Cache)
|
||||
m.Bubblewrap = ProbeBubblewrap
|
||||
// Outside a launch by the runtime — a build, a check — there is no node and no state to write.
|
||||
|
||||
@@ -82,6 +82,10 @@ type Module struct {
|
||||
Settings string
|
||||
Capacity func() int
|
||||
ReadKept func(key string) (json.RawMessage, error)
|
||||
// ReadManifest reads a key of the delivery indexer's manifests state (ManifestsState), which a compose ask's request
|
||||
// names its manifests by (compose.go); GoModCache, when set, is the Go module cache in place of Cache's (a test).
|
||||
ReadManifest func(key string) (json.RawMessage, error)
|
||||
GoModCache string
|
||||
|
||||
spacesOnce sync.Once
|
||||
workspaces *buildstep.Workspaces
|
||||
|
||||
@@ -0,0 +1,79 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"reflect"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/buildstep"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
)
|
||||
|
||||
// 1925/AC1: packagesArg takes the list as the bus's JSON shape ([]any) and as the typed list an ask hands in process
|
||||
// ([]string), and refuses a bad element in either.
|
||||
func TestPackagesArgTakesBothShapes(t *testing.T) {
|
||||
cases := []struct {
|
||||
name string
|
||||
in any
|
||||
want []string
|
||||
refused bool
|
||||
}{
|
||||
{"absent", nil, nil, false},
|
||||
{"typed", []string{"x/a", "x/b/..."}, []string{"x/a", "x/b/..."}, false},
|
||||
{"json", []any{"x/a", "x/b/..."}, []string{"x/a", "x/b/..."}, false},
|
||||
{"typed empty", []string{}, []string{}, false},
|
||||
{"typed dotdot", []string{"x/../y"}, nil, true},
|
||||
{"json dotdot", []any{"x/../y"}, nil, true},
|
||||
{"typed bad", []string{"x a"}, nil, true},
|
||||
{"json bad", []any{"x a"}, nil, true},
|
||||
{"json not a string", []any{3}, nil, true},
|
||||
{"a string", "x/a", nil, true},
|
||||
}
|
||||
for _, c := range cases {
|
||||
got, err := packagesArg(c.in)
|
||||
if c.refused != (err != nil) {
|
||||
t.Errorf("%s: err = %v, refused want %v", c.name, err, c.refused)
|
||||
continue
|
||||
}
|
||||
if !c.refused && !reflect.DeepEqual(got, c.want) {
|
||||
t.Errorf("%s: got %#v, want %#v", c.name, got, c.want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// 1925/AC2: a check ask through the module's real build_go_test body gets past the packages argument: the digest names
|
||||
// a package, buildstep.RunAsk hands it in as a typed list, and the step fails only later, on the workspace nobody fetched.
|
||||
func TestACheckAskReachesTheTestStepPastItsPackages(t *testing.T) {
|
||||
m := &Module{Node: "ace", Work: t.TempDir()}
|
||||
var calls []string
|
||||
steps := fakeSteps(&calls)
|
||||
for i := range steps {
|
||||
switch steps[i].Name {
|
||||
case "build_go_digest":
|
||||
steps[i].Run = func(ctx context.Context, args map[string]any) (any, error) {
|
||||
return map[string]any{"level": "unit", "packages": map[string]string{"x/a": "d1"}}, nil
|
||||
}
|
||||
case "build_go_test":
|
||||
for _, real := range m.StepTools() {
|
||||
if real.Name == "build_go_test" {
|
||||
steps[i].Run = real.Run
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
a := delivery.StepAsk{ID: buildstep.NewID(), Kind: delivery.AskCheck, Language: delivery.AskGo, Asker: "mesh-delivery-checker",
|
||||
At: time.Now(), Check: &delivery.CheckJob{Fetch: map[string]any{"repository": "novox/x", "commit": askCommit}, Levels: []string{delivery.LevelUnit}}}
|
||||
out, err := buildstep.RunAsk(context.Background(), m.askSteps(steps), a)
|
||||
got := ""
|
||||
if err != nil {
|
||||
got = err.Error()
|
||||
}
|
||||
if strings.Contains(got, "packages is a list") {
|
||||
t.Fatalf("the test step refused the ask's packages: %s", got)
|
||||
}
|
||||
if !strings.Contains(got, "no workspace") {
|
||||
t.Fatalf("want the step to fail on the unfetched workspace, past its packages; err=%v out=%+v", err, out)
|
||||
}
|
||||
}
|
||||
@@ -127,17 +127,25 @@ func (m *Module) besideEnv(workspace string) []string {
|
||||
return []string{"MESH_CHECK_BESIDE=" + dir}
|
||||
}
|
||||
|
||||
// packagesArg reads the packages argument: a list of package patterns, as []any when it came over the bus as JSON and
|
||||
// as []string when an ask handed it in process (issue 1925).
|
||||
func packagesArg(v any) ([]string, error) {
|
||||
if v == nil {
|
||||
var list []string
|
||||
switch t := v.(type) {
|
||||
case nil:
|
||||
return nil, nil
|
||||
}
|
||||
list, ok := v.([]any)
|
||||
if !ok {
|
||||
case []string:
|
||||
list = t
|
||||
case []any:
|
||||
for _, e := range t {
|
||||
s, _ := e.(string)
|
||||
list = append(list, s)
|
||||
}
|
||||
default:
|
||||
return nil, fmt.Errorf("packages is a list of package patterns")
|
||||
}
|
||||
out := []string{}
|
||||
for _, e := range list {
|
||||
s, _ := e.(string)
|
||||
for _, s := range list {
|
||||
if !packageOK.MatchString(s) || strings.Contains(strings.TrimSuffix(s, "/..."), "..") {
|
||||
return nil, fmt.Errorf("%q is not a package pattern", s)
|
||||
}
|
||||
@@ -201,13 +209,11 @@ func (m *Module) runLevel(ctx context.Context, tree string, o levelOpts) (*TestR
|
||||
// gofmt and go vet of each directory the level runs come before its tests (ADR 0529 §1; issue 1887 D7-go): a
|
||||
// directory that fails them fails the level at once, and their time is the result's Lint.
|
||||
linted := m.clock()
|
||||
var failed []Failure
|
||||
for _, d := range todo {
|
||||
f, err := m.lint(ctx, run, tree, d, append(m.goEnvFor(filepath.Join(tree, d)), o.env...))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
failed = append(failed, f...)
|
||||
failed, err := m.lintAll(ctx, run, tree, todo, func(d string) []string {
|
||||
return append(m.goEnvFor(filepath.Join(tree, d)), o.env...)
|
||||
})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
res.Lint = m.clock().Sub(linted).Seconds()
|
||||
if len(failed) > 0 {
|
||||
|
||||
Vendored
+13
@@ -0,0 +1,13 @@
|
||||
{"Action":"start","Package":"x"}
|
||||
{"Action":"run","Package":"x","Test":"TestThree"}
|
||||
{"Action":"output","Package":"x","Test":"TestThree","Output":"=== RUN TestThree\n","OutputType":"frame"}
|
||||
{"Action":"output","Package":"x","Test":"TestThree","Output":" a_test.go:4: judged against /var/lib/mesh-build-go-work/id/beside/mesh-catalog as cloned\n"}
|
||||
{"Action":"output","Package":"x","Test":"TestThree","Output":" a_test.go:5: 3 subscription(s) derive to a subject no emitter owns:\n","OutputType":"error"}
|
||||
{"Action":"output","Package":"x","Test":"TestThree","Output":" a.b.c\n","OutputType":"error-continue"}
|
||||
{"Action":"output","Package":"x","Test":"TestThree","Output":" d.e.f\n","OutputType":"error-continue"}
|
||||
{"Action":"output","Package":"x","Test":"TestThree","Output":" g.h.i\n","OutputType":"error-continue"}
|
||||
{"Action":"output","Package":"x","Test":"TestThree","Output":"--- FAIL: TestThree (0.00s)\n","OutputType":"frame"}
|
||||
{"Action":"fail","Package":"x","Test":"TestThree","Elapsed":0}
|
||||
{"Action":"output","Package":"x","Output":"FAIL\n","OutputType":"frame"}
|
||||
{"Action":"output","Package":"x","Output":"FAIL\tx\t0.002s\n","OutputType":"frame"}
|
||||
{"Action":"fail","Package":"x","Elapsed":0.002}
|
||||
@@ -3,7 +3,7 @@ module meshbuildgo
|
||||
go 1.26.0
|
||||
|
||||
require (
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023150321-a2c813a6b468
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024021031-66bf26c0d721
|
||||
github.com/distribution/distribution/v3 v3.1.2
|
||||
github.com/nats-io/nats-server/v2 v2.11.17
|
||||
github.com/nats-io/nats.go v1.54.0
|
||||
|
||||
@@ -1,5 +1,7 @@
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023150321-a2c813a6b468 h1:UcMee+IV2ii5E4dRZdQUaSaeh9IedTFYumuQbBiB0k0=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023150321-a2c813a6b468/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023213200-403a5d72ab93 h1:FAiFY2KBKEoZSH49OZTt7/zRBIQxyl5g/KhYrpggbN4=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023213200-403a5d72ab93/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024021031-66bf26c0d721 h1:N//dDGmry1F3HKeBTWd+z/SfK+wb8bUUMJF0OlgR0/g=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024021031-66bf26c0d721/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
|
||||
github.com/alicebob/miniredis/v2 v2.35.0 h1:QwLphYqCEAo1eu1TqPRN2jgVMPBweeQcR21jeqDCONI=
|
||||
github.com/alicebob/miniredis/v2 v2.35.0/go.mod h1:TcL7YfarKPGDAthEtl5NBeHZfeUQj6OXMm/+iu5cLMM=
|
||||
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0=
|
||||
|
||||
@@ -20,7 +20,8 @@
|
||||
}
|
||||
],
|
||||
"reads": [
|
||||
"mesh-delivery-checker.kept"
|
||||
"mesh-delivery-checker.kept",
|
||||
"mesh-delivery-indexer.manifests"
|
||||
],
|
||||
"config": {
|
||||
"capacity": {
|
||||
|
||||
@@ -3,7 +3,7 @@ module meshbuildimage
|
||||
go 1.26.0
|
||||
|
||||
require (
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023150321-a2c813a6b468
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023213200-403a5d72ab93
|
||||
github.com/nats-io/nats-server/v2 v2.11.17
|
||||
github.com/nats-io/nats.go v1.54.0
|
||||
)
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023150321-a2c813a6b468 h1:UcMee+IV2ii5E4dRZdQUaSaeh9IedTFYumuQbBiB0k0=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023150321-a2c813a6b468/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023213200-403a5d72ab93 h1:FAiFY2KBKEoZSH49OZTt7/zRBIQxyl5g/KhYrpggbN4=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023213200-403a5d72ab93/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
|
||||
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0=
|
||||
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI=
|
||||
github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=
|
||||
|
||||
@@ -3,7 +3,7 @@ module meshbuildnode
|
||||
go 1.26.0
|
||||
|
||||
require (
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023150321-a2c813a6b468
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023213200-403a5d72ab93
|
||||
github.com/distribution/distribution/v3 v3.1.2
|
||||
github.com/nats-io/nats.go v1.54.0
|
||||
)
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023150321-a2c813a6b468 h1:UcMee+IV2ii5E4dRZdQUaSaeh9IedTFYumuQbBiB0k0=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023150321-a2c813a6b468/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023213200-403a5d72ab93 h1:FAiFY2KBKEoZSH49OZTt7/zRBIQxyl5g/KhYrpggbN4=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023213200-403a5d72ab93/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
|
||||
github.com/alicebob/miniredis/v2 v2.35.0 h1:QwLphYqCEAo1eu1TqPRN2jgVMPBweeQcR21jeqDCONI=
|
||||
github.com/alicebob/miniredis/v2 v2.35.0/go.mod h1:TcL7YfarKPGDAthEtl5NBeHZfeUQj6OXMm/+iu5cLMM=
|
||||
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0=
|
||||
|
||||
@@ -514,6 +514,11 @@ func TestAPassingVerdictAsksTheApproveAndItsApproveMergesWithinSecondsOverARealB
|
||||
announced.Title = "Fix the apostrophe (issue 975)"
|
||||
announced.Body = "The description, with no review line yet."
|
||||
b.publish(t, "mesh.mod.gitea.event.pull.updated", announced)
|
||||
rt.until("the pull request's key to hold its head", eventWithin, func() bool {
|
||||
e, err := b.kvs["pulls"].Get(ctx, "novox.app.7")
|
||||
var k PullKey
|
||||
return err == nil && json.Unmarshal(e.Value(), &k) == nil && k.Head == head && k.Trunk == "a1a1a1a1a1a1"
|
||||
})
|
||||
b.checkPasses(t, rt, "fix/x", "a1a1a1a1a1a1", judgedPull{"novox/app", 7, head})
|
||||
|
||||
// No review names the head, so nothing is asked (ADR 0496, issue 1803): the delivery waits for a review.
|
||||
|
||||
@@ -51,7 +51,38 @@ func (h *Holder) KeepPull(p PullEvent) {
|
||||
if err != nil {
|
||||
h.Logf("[mesh-delivery] the key of %s/%s#%d could not be written: %v", p.Owner, p.Repo, p.Number, err)
|
||||
h.again("a pull request's key that could not be written")
|
||||
return
|
||||
}
|
||||
// The head's mesh/check may have been heard before this key held the head: the two events are taken on goroutines of
|
||||
// their own, and KeepCheckVerdict then found no head to write a verdict to. The verdict the delivery holds for this head
|
||||
// and trunk is written now; read after the write above landed, it cannot miss one taken before that write.
|
||||
verdict := h.checkVerdictOf(p.repository(), p.HeadSHA, p.BaseSHA)
|
||||
if verdict == "" {
|
||||
return
|
||||
}
|
||||
err = h.Pulls.Change(pullKeyName(p.Owner, p.Repo, p.Number), func(k PullKey) (PullKey, bool) {
|
||||
if !sameCommit(k.Head, p.HeadSHA) || !sameCommit(k.Trunk, p.BaseSHA) || k.Verdict != "" {
|
||||
return k, false
|
||||
}
|
||||
k.Verdict = verdict
|
||||
return k, true
|
||||
})
|
||||
if err != nil {
|
||||
h.Logf("[mesh-delivery] the verdict in the key of %s/%s#%d could not be written: %v", p.Owner, p.Repo, p.Number, err)
|
||||
h.again("a pull request's key that could not be written")
|
||||
}
|
||||
}
|
||||
|
||||
// checkVerdictOf is the verdict of the mesh/check the delivery of a head holds, when it judged this head on this trunk
|
||||
// commit; empty otherwise. Outside the lock.
|
||||
func (h *Holder) checkVerdictOf(repository, head, trunk string) string {
|
||||
h.mu.Lock()
|
||||
defer h.mu.Unlock()
|
||||
d := h.deliveries[IDOf(strings.ToLower(repository), head)]
|
||||
if d == nil || d.Check == nil || d.Check.Status != ctxCheck || !sameCommit(d.Check.Head, head) || !sameCommit(d.Check.Trunk, trunk) {
|
||||
return ""
|
||||
}
|
||||
return d.Check.Gate
|
||||
}
|
||||
|
||||
// KeepCheckVerdict writes the verdict of a head's mesh/check into the key of its pull request, when the key holds that head
|
||||
|
||||
@@ -145,3 +145,21 @@ func TestTheSameMergeHeardAgainAfterALaterMergeMovesNoTrunkBack(t *testing.T) {
|
||||
t.Fatalf("the trunk moved from %q to %q on a merge heard again; wanted it to stay m9m9m9m9m9m9", before.TrunkNow, after.TrunkNow)
|
||||
}
|
||||
}
|
||||
|
||||
// The forge's mesh/check event handled wholly between the announcement's PullUpdated and its KeepPull (the two events are
|
||||
// taken on goroutines of their own): the verdict must still reach the pull request's key.
|
||||
func TestAVerdictHeardBeforeTheKeyIsWrittenReachesTheKey(t *testing.T) {
|
||||
w := newWorld(t)
|
||||
pulls := newFakePulls()
|
||||
w.h.Pulls = pulls
|
||||
p := announcedAt(readyHead, "a1a1a1a1a1a1")
|
||||
w.h.PullUpdated(p)
|
||||
s := ForgeStatusEvent{Owner: "novox", Repo: "app", SHA: readyHead, Context: ctxCheck, State: "success",
|
||||
Description: "pass " + readyHead[:12] + " on a1a1a1a1a1a1 | unit 3 run"}
|
||||
w.h.CommitStatus(s)
|
||||
w.h.KeepCheckVerdict(s)
|
||||
w.h.KeepPull(p)
|
||||
if got := pulls.get(appKey); got.Head != readyHead || got.Verdict != "pass" {
|
||||
t.Fatalf("the key holds %+v, want the head with the verdict pass", got)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -90,7 +90,9 @@ func TestAcceptancePublishedMailBecomesAnEntryNeverKeepingItsBody(t *testing.T)
|
||||
}
|
||||
r.stand.Judge = judgeEveryMailAs("neither", "newsletters", "", "")
|
||||
r.stand.MessengerCheck = func(map[string]any) (any, error) { return map[string]any{"allowed": true}, nil }
|
||||
r.stand.Tell = func(map[string]any) (any, error) { return map[string]any{"sent_on": ""}, nil }
|
||||
r.stand.Tell = func(map[string]any) (any, error) {
|
||||
return map[string]any{"sent_on": nil, "dropped": []string{"telegram"}}, nil
|
||||
}
|
||||
|
||||
publishAdded(t, r.nc, "bob", "INBOX", 1, []map[string]any{{"name": "Alice", "address": "alice@example.org"}}, "hi", time.Now())
|
||||
|
||||
@@ -160,7 +162,7 @@ func TestAcceptanceOneMatteringMailSendsExactlyOneTellNamingIt(t *testing.T) {
|
||||
var tells recorded[map[string]any]
|
||||
r.stand.Tell = func(args map[string]any) (any, error) {
|
||||
tells.add(args)
|
||||
return map[string]any{"sent_on": time.Now().UTC().Format(time.RFC3339)}, nil
|
||||
return map[string]any{"sent_on": []string{"telegram"}, "dropped": []string{}}, nil
|
||||
}
|
||||
|
||||
publishAdded(t, r.nc, "bob", "INBOX", 1, []map[string]any{{"name": "Alice", "address": "alice@example.org"}}, "invoice", time.Now())
|
||||
@@ -208,7 +210,7 @@ func TestAcceptanceSenderNameWithAHostNameGivesTheSafeForm(t *testing.T) {
|
||||
var tells recorded[map[string]any]
|
||||
r.stand.Tell = func(args map[string]any) (any, error) {
|
||||
tells.add(args)
|
||||
return map[string]any{"sent_on": time.Now().UTC().Format(time.RFC3339)}, nil
|
||||
return map[string]any{"sent_on": []string{"telegram"}, "dropped": []string{}}, nil
|
||||
}
|
||||
|
||||
publishAdded(t, r.nc, "bob", "INBOX", 1, []map[string]any{{"name": "mail.evil.example.com", "address": "x@example.org"}}, "hi", time.Now())
|
||||
@@ -233,9 +235,9 @@ func TestAcceptanceDroppedTellLeavesMailUntoldAndCountedNextTime(t *testing.T) {
|
||||
var tellCalls count
|
||||
r.stand.Tell = func(map[string]any) (any, error) {
|
||||
if tellCalls.inc() == 1 {
|
||||
return map[string]any{"sent_on": ""}, nil // dropped: the operator channel's own cap
|
||||
return map[string]any{"sent_on": nil, "dropped": []string{"telegram"}}, nil // dropped: the operator channel's own cap
|
||||
}
|
||||
return map[string]any{"sent_on": time.Now().UTC().Format(time.RFC3339)}, nil
|
||||
return map[string]any{"sent_on": []string{"telegram"}, "dropped": []string{}}, nil
|
||||
}
|
||||
|
||||
publishAdded(t, r.nc, "bob", "INBOX", 1, []map[string]any{{"name": "Alice", "address": "alice@example.org"}}, "hi", time.Now())
|
||||
@@ -269,3 +271,38 @@ func TestAcceptanceDroppedTellLeavesMailUntoldAndCountedNextTime(t *testing.T) {
|
||||
})
|
||||
assertNeverSendsMovesFlagsOrDeletes(t, r)
|
||||
}
|
||||
|
||||
// 1926/AC4: a second batch over an unchanged set of mails that matter (its own mail does not matter) sends no
|
||||
// second signal — the first, answered as the real operator-channel answers it (sent_on a list), marked the set
|
||||
// told.
|
||||
func TestAcceptanceRepeatedBatchesOverAnUnchangedSetSendOneSignal(t *testing.T) {
|
||||
r := startReader(t)
|
||||
r.stand.MailAccounts = oneAccountAnswer("bob")
|
||||
r.stand.MailRead = func(map[string]any) (any, error) { return map[string]any{"text": "hi"}, nil }
|
||||
var judged count
|
||||
matters := judgeEveryMailAs("matters", "", "", "")
|
||||
neither := judgeEveryMailAs("neither", "", "", "")
|
||||
r.stand.Judge = func(args map[string]any) (any, error) {
|
||||
if judged.inc() == 1 {
|
||||
return matters(args)
|
||||
}
|
||||
return neither(args)
|
||||
}
|
||||
r.stand.MessengerCheck = func(map[string]any) (any, error) { return map[string]any{"allowed": true}, nil }
|
||||
var tells count
|
||||
r.stand.Tell = func(map[string]any) (any, error) {
|
||||
tells.inc()
|
||||
return map[string]any{"sent_on": []string{"telegram"}, "dropped": []string{}}, nil
|
||||
}
|
||||
|
||||
publishAdded(t, r.nc, "bob", "INBOX", 1, []map[string]any{{"name": "Alice", "address": "alice@example.org"}}, "hi", time.Now())
|
||||
waitFor(t, 30*time.Second, "the first tell", func() bool { return tells.get() >= 1 })
|
||||
publishAdded(t, r.nc, "bob", "INBOX", 2, []map[string]any{{"name": "Bob", "address": "bob2@example.org"}}, "second", time.Now())
|
||||
publishAdded(t, r.nc, "bob", "INBOX", 3, []map[string]any{{"name": "Carol", "address": "carol@example.org"}}, "third", time.Now())
|
||||
waitFor(t, BatchGap+60*time.Second, "later batches judged", func() bool { return judged.get() >= 2 })
|
||||
time.Sleep(2 * BatchGap) // a stray repeat of the signal would come after a batch
|
||||
if got := tells.get(); got != 1 {
|
||||
t.Fatalf("1926/AC4: tells = %d, want exactly 1 for an unchanged set", got)
|
||||
}
|
||||
assertNeverSendsMovesFlagsOrDeletes(t, r)
|
||||
}
|
||||
|
||||
@@ -321,7 +321,7 @@ func RunBatches(ctx context.Context, store Store, readText ReadTextFunc, judge J
|
||||
if err := RunOneBatch(ctx, store, readText, judge, waiting, now, logf); err != nil {
|
||||
logf("[mesh-mail-reader] a batch failed: %v", err)
|
||||
}
|
||||
if err := SendSignal(ctx, store, check, tell); err != nil {
|
||||
if err := SendSignal(ctx, store, check, tell, now); err != nil {
|
||||
logf("[mesh-mail-reader] the mail signal could not be sent: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -58,9 +58,9 @@ func TestAcceptanceRefusedJudgementIsUnhealthyThenCatchesUpInPublishedOrder(t *t
|
||||
var tellCalls count
|
||||
r.stand.Tell = func(map[string]any) (any, error) {
|
||||
if tellCalls.inc() == 1 {
|
||||
return map[string]any{"sent_on": ""}, nil // dropped: the operator channel's own cap (1522/AC2)
|
||||
return map[string]any{"sent_on": nil, "dropped": []string{"telegram"}}, nil // dropped: the operator channel's own cap (1522/AC2)
|
||||
}
|
||||
return map[string]any{"sent_on": time.Now().UTC().Format(time.RFC3339)}, nil
|
||||
return map[string]any{"sent_on": []string{"telegram"}, "dropped": []string{}}, nil
|
||||
}
|
||||
|
||||
var mu sync.Mutex
|
||||
|
||||
@@ -8,6 +8,8 @@ import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
||||
// MessengerCheck is one seat:messenger.messenger_check ask (ADR 0461 §4): whether text is safe to say on an
|
||||
@@ -44,15 +46,75 @@ func NewTell(ask Ask) Tell {
|
||||
return "", err
|
||||
}
|
||||
var out struct {
|
||||
SentOn string `json:"sent_on"`
|
||||
SentOn json.RawMessage `json:"sent_on"`
|
||||
}
|
||||
if err := json.Unmarshal(raw, &out); err != nil {
|
||||
return "", fmt.Errorf("operator-channel.tell's answer could not be read: %.200s", raw)
|
||||
}
|
||||
return out.SentOn, nil
|
||||
sentOn, err := readSentOn(out.SentOn)
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("operator-channel.tell's answer could not be read: %.200s", raw)
|
||||
}
|
||||
return sentOn, nil
|
||||
}
|
||||
}
|
||||
|
||||
// readSentOn reads tell's sent_on: messenger answers the list of channels that said it (null or empty when
|
||||
// every channel dropped it), never one string (issue 1926: reading it as a string failed every real answer, so
|
||||
// no mail was ever marked told and the same signal went out after every batch). A plain string is still
|
||||
// accepted. The channels are joined with a comma; empty means nothing said it.
|
||||
func readSentOn(raw json.RawMessage) (string, error) {
|
||||
if len(raw) == 0 || string(raw) == "null" {
|
||||
return "", nil
|
||||
}
|
||||
var list []string
|
||||
if err := json.Unmarshal(raw, &list); err == nil {
|
||||
return strings.Join(list, ","), nil
|
||||
}
|
||||
var one string
|
||||
if err := json.Unmarshal(raw, &one); err != nil {
|
||||
return "", err
|
||||
}
|
||||
return one, nil
|
||||
}
|
||||
|
||||
// signalEntries is what the signal counts out of the entries marked matters and not yet told (issue 1926):
|
||||
// a mail whose due date lies before today is left out (AC2: it no longer matters, the mail page still holds
|
||||
// it); the same mail seen twice (one message id in two folders, issue 1849) and the same notice in two
|
||||
// languages (the same sender address, mail day and due date) count once (AC3). The entries left out stay
|
||||
// as they were; the ones that count are returned in the order given.
|
||||
func signalEntries(entries []Entry, now time.Time) []Entry {
|
||||
today := time.Date(now.UTC().Year(), now.UTC().Month(), now.UTC().Day(), 0, 0, 0, 0, time.UTC)
|
||||
seen := map[string]bool{}
|
||||
var out []Entry
|
||||
for _, e := range entries {
|
||||
if !e.DueDate.IsZero() && e.DueDate.UTC().Before(today) {
|
||||
continue
|
||||
}
|
||||
var keys []string
|
||||
if e.MessageID != "" {
|
||||
keys = append(keys, "id\x00"+e.Account+"\x00"+e.MessageID)
|
||||
}
|
||||
if !e.DueDate.IsZero() && e.SenderAddress != "" {
|
||||
keys = append(keys, "notice\x00"+e.Account+"\x00"+strings.ToLower(e.SenderAddress)+"\x00"+
|
||||
e.Date.UTC().Format("2006-01-02")+"\x00"+e.DueDate.UTC().Format("2006-01-02"))
|
||||
}
|
||||
dup := false
|
||||
for _, k := range keys {
|
||||
if seen[k] {
|
||||
dup = true
|
||||
}
|
||||
}
|
||||
for _, k := range keys {
|
||||
seen[k] = true
|
||||
}
|
||||
if !dup {
|
||||
out = append(out, e)
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// SendSignal is 1522: every entry marked matters and not yet told — including one carried over from an
|
||||
// earlier batch that went untold (1429/AC4) — is gathered; its words are built (signal.go) and checked word
|
||||
// by word with check; any one refused falls back to the safe form (1522/AC1). Nothing is sent, and nothing
|
||||
@@ -61,7 +123,7 @@ func NewTell(ask Ask) Tell {
|
||||
// own last outcome. The tell sent, every one of those entries is marked told only once it answers sent_on
|
||||
// (1522/AC2); a tell that answers sent_on empty leaves them exactly as they were, for the next signal to
|
||||
// count (1522/AC2, AC3).
|
||||
func SendSignal(ctx context.Context, store Store, check MessengerCheck, tell Tell) error {
|
||||
func SendSignal(ctx context.Context, store Store, check MessengerCheck, tell Tell, now func() time.Time) error {
|
||||
health, err := store.ReaderHealth(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
@@ -73,17 +135,23 @@ func SendSignal(ctx context.Context, store Store, check MessengerCheck, tell Tel
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
var toTell []Entry
|
||||
var mails []SignalMail
|
||||
var untold []Entry
|
||||
for _, e := range entries {
|
||||
if e.Mark == MarkMatters && !e.Told {
|
||||
toTell = append(toTell, e)
|
||||
mails = append(mails, SignalMail{SenderName: e.SenderName, Summary: e.Summary, Date: e.Date, DueDate: e.DueDate})
|
||||
untold = append(untold, e)
|
||||
}
|
||||
}
|
||||
// What counts (signalEntries) is what is said; every untold entry is marked told once it is said, so an
|
||||
// unchanged set is never said twice (issue 1926/AC1) and one that only holds past-due or duplicate mails
|
||||
// says nothing.
|
||||
toTell := signalEntries(untold, now())
|
||||
if len(toTell) == 0 {
|
||||
return nil
|
||||
}
|
||||
var mails []SignalMail
|
||||
for _, e := range toTell {
|
||||
mails = append(mails, SignalMail{SenderName: e.SenderName, Summary: e.Summary, Date: e.Date, DueDate: e.DueDate})
|
||||
}
|
||||
|
||||
words := BuildSignal(mails)
|
||||
allowed := true
|
||||
@@ -108,7 +176,7 @@ func SendSignal(ctx context.Context, store Store, check MessengerCheck, tell Tel
|
||||
if sentOn == "" {
|
||||
return nil
|
||||
}
|
||||
for _, e := range toTell {
|
||||
for _, e := range untold {
|
||||
e.Told = true
|
||||
if err := store.UpdateEntry(ctx, e); err != nil {
|
||||
return err
|
||||
|
||||
@@ -2,6 +2,7 @@ package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
@@ -38,7 +39,7 @@ func TestSendSignalChecksWordsThenTellsWithSubjectMail(t *testing.T) {
|
||||
check, asked := recordingCheck(true)
|
||||
tell, headlines, subjects := recordingTell("2026-10-21T10:00:00Z")
|
||||
|
||||
if err := SendSignal(ctx, store, check, tell); err != nil {
|
||||
if err := SendSignal(ctx, store, check, tell, time.Now); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(*asked) == 0 || (*asked)[0] != "1 mail matters" {
|
||||
@@ -64,7 +65,7 @@ func TestSendSignalFallsBackToTheSafeFormWhenRefused(t *testing.T) {
|
||||
check, _ := recordingCheck(false)
|
||||
tell, headlines, _ := recordingTell("sent")
|
||||
|
||||
if err := SendSignal(ctx, store, check, tell); err != nil {
|
||||
if err := SendSignal(ctx, store, check, tell, time.Now); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(*headlines) != 1 {
|
||||
@@ -81,7 +82,7 @@ func TestSendSignalSentOnEmptyLeavesMailsUntold(t *testing.T) {
|
||||
check, _ := recordingCheck(true)
|
||||
tellDropped, calls, _ := recordingTell("") // every channel dropped it
|
||||
|
||||
if err := SendSignal(ctx, store, check, tellDropped); err != nil {
|
||||
if err := SendSignal(ctx, store, check, tellDropped, time.Now); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
entry, _, _ := store.Entry(ctx, id)
|
||||
@@ -94,7 +95,7 @@ func TestSendSignalSentOnEmptyLeavesMailsUntold(t *testing.T) {
|
||||
|
||||
// The next signal counts the still-untold mail (1429/AC4): SendSignal asks about it again.
|
||||
tellSent, headlines, _ := recordingTell("sent-now")
|
||||
if err := SendSignal(ctx, store, check, tellSent); err != nil {
|
||||
if err := SendSignal(ctx, store, check, tellSent, time.Now); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(*headlines) != 1 {
|
||||
@@ -114,7 +115,7 @@ func TestSendSignalSendsNothingForScamWarningsAlone(t *testing.T) {
|
||||
check, asked := recordingCheck(true)
|
||||
tell, headlines, _ := recordingTell("sent")
|
||||
|
||||
if err := SendSignal(ctx, store, check, tell); err != nil {
|
||||
if err := SendSignal(ctx, store, check, tell, time.Now); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(*asked) != 0 || len(*headlines) != 0 {
|
||||
@@ -132,7 +133,7 @@ func TestSendSignalSendsNothingWhileUnhealthy(t *testing.T) {
|
||||
check, asked := recordingCheck(true)
|
||||
tell, headlines, _ := recordingTell("sent")
|
||||
|
||||
if err := SendSignal(ctx, store, check, tell); err != nil {
|
||||
if err := SendSignal(ctx, store, check, tell, time.Now); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(*asked) != 0 || len(*headlines) != 0 {
|
||||
@@ -141,7 +142,7 @@ func TestSendSignalSendsNothingWhileUnhealthy(t *testing.T) {
|
||||
|
||||
// After recovery, the first signal counts the mail that went untold while unhealthy.
|
||||
store.SetReaderHealth(ctx, ReaderHealth{Healthy: true})
|
||||
if err := SendSignal(ctx, store, check, tell); err != nil {
|
||||
if err := SendSignal(ctx, store, check, tell, time.Now); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(*headlines) != 1 {
|
||||
@@ -166,3 +167,120 @@ func TestTheManifestInvokesTellAndMessengerCheck(t *testing.T) {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
var signalNow = func() time.Time { return time.Date(2026, 10, 23, 21, 0, 0, 0, time.UTC) }
|
||||
|
||||
// 1926/AC1: the real operator-channel.tell answers sent_on as the list of channels that said it (messenger's
|
||||
// own shape); read as a string it failed, nothing was marked told and every batch sent the same signal again.
|
||||
func TestNewTellReadsSentOnAsMessengerAnswersIt(t *testing.T) {
|
||||
for name, answer := range map[string]string{
|
||||
"a list": `{"key":"tell.1","sent_on":["telegram"],"dropped":[]}`,
|
||||
"a string": `{"sent_on":"telegram"}`,
|
||||
} {
|
||||
tell := NewTell(func(string, map[string]any) (json.RawMessage, error) { return json.RawMessage(answer), nil })
|
||||
sentOn, err := tell("h", nil, "mail")
|
||||
if err != nil || sentOn == "" {
|
||||
t.Fatalf("1926/AC1: %s: sentOn = %q, err = %v, want a channel", name, sentOn, err)
|
||||
}
|
||||
}
|
||||
for name, answer := range map[string]string{
|
||||
"null": `{"sent_on":null,"dropped":["telegram"]}`,
|
||||
"empty list": `{"sent_on":[],"dropped":["telegram"]}`,
|
||||
"absent": `{"dropped":["telegram"]}`,
|
||||
} {
|
||||
tell := NewTell(func(string, map[string]any) (json.RawMessage, error) { return json.RawMessage(answer), nil })
|
||||
sentOn, err := tell("h", nil, "mail")
|
||||
if err != nil || sentOn != "" {
|
||||
t.Fatalf("1926/AC1: %s: sentOn = %q, err = %v, want empty and no error", name, sentOn, err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// 1926/AC1: a set already told is not sent again; a mail added to it is sent alone.
|
||||
func TestSendSignalSendsAnUnchangedSetOnce(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
store := newMemStore()
|
||||
for i := 0; i < 3; i++ {
|
||||
store.AddEntry(ctx, mattersEntry("Alice", "pay the bill"))
|
||||
}
|
||||
check, _ := recordingCheck(true)
|
||||
tell, headlines, _ := recordingTell("telegram")
|
||||
for i := 0; i < 3; i++ {
|
||||
if err := SendSignal(ctx, store, check, tell, signalNow); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
if len(*headlines) != 1 || (*headlines)[0] != "3 mails matter" {
|
||||
t.Fatalf("1926/AC1: tell headlines = %v, want one \"3 mails matter\"", *headlines)
|
||||
}
|
||||
store.AddEntry(ctx, mattersEntry("Bob", "new"))
|
||||
if err := SendSignal(ctx, store, check, tell, signalNow); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(*headlines) != 2 || (*headlines)[1] != "1 mail matters" {
|
||||
t.Fatalf("1926/AC1: tell headlines = %v, want the changed set sent again", *headlines)
|
||||
}
|
||||
}
|
||||
|
||||
// 1926/AC2: a mail whose due date has passed is not counted and does not lead; one due today still leads.
|
||||
func TestSendSignalLeavesPastDueMailsOut(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
store := newMemStore()
|
||||
past := mattersEntry("bpost", "parcel notice")
|
||||
past.DueDate = time.Date(2026, 10, 22, 0, 0, 0, 0, time.UTC)
|
||||
today := mattersEntry("Alice", "pay today")
|
||||
today.DueDate = time.Date(2026, 10, 23, 0, 0, 0, 0, time.UTC)
|
||||
store.AddEntry(ctx, past)
|
||||
store.AddEntry(ctx, today)
|
||||
check, _ := recordingCheck(true)
|
||||
var headlines []string
|
||||
var lines [][]string
|
||||
tell := func(h string, l []string, _ string) (string, error) {
|
||||
headlines, lines = append(headlines, h), append(lines, l)
|
||||
return "telegram", nil
|
||||
}
|
||||
if err := SendSignal(ctx, store, check, tell, signalNow); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(headlines) != 1 || headlines[0] != "1 mail matters" || len(lines[0]) != 1 || lines[0][0] != "Alice — pay today — due 2026-10-23" {
|
||||
t.Fatalf("1926/AC2: told %v %v, want only Alice's mail due today", headlines, lines)
|
||||
}
|
||||
|
||||
// A set holding only past-due mails says nothing.
|
||||
store2 := newMemStore()
|
||||
store2.AddEntry(ctx, past)
|
||||
headlines = nil
|
||||
if err := SendSignal(ctx, store2, check, tell, signalNow); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(headlines) != 0 {
|
||||
t.Fatalf("1926/AC2: told %v for past-due mails alone, want nothing", headlines)
|
||||
}
|
||||
}
|
||||
|
||||
// 1926/AC3: the same notice in two languages (same sender, same day, same due date) counts once, and so does
|
||||
// the same message id in two folders (issue 1849); two different mails of one sender stay two.
|
||||
func TestSendSignalCountsADuplicateNoticeOnce(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
store := newMemStore()
|
||||
due := time.Date(2026, 10, 30, 0, 0, 0, 0, time.UTC)
|
||||
day := time.Date(2026, 10, 21, 8, 0, 0, 0, time.UTC)
|
||||
en := Entry{Account: "bob", MessageID: "<a>", SenderName: "bpost", SenderAddress: "no-reply@bpost.example", Summary: "parcel waiting", Mark: MarkMatters, Date: day, DueDate: due}
|
||||
nl := en
|
||||
nl.MessageID, nl.Summary, nl.Date = "<b>", "pakje wacht", day.Add(time.Minute)
|
||||
inbox := Entry{Account: "bob", MessageID: "<c>", Folder: "INBOX", SenderName: "Carol", SenderAddress: "carol@x.example", Summary: "hello", Mark: MarkMatters, Date: day}
|
||||
allMail := inbox
|
||||
allMail.Folder = "All Mail"
|
||||
other := Entry{Account: "bob", MessageID: "<d>", SenderName: "Carol", SenderAddress: "carol@x.example", Summary: "second", Mark: MarkMatters, Date: day}
|
||||
for _, e := range []Entry{en, nl, inbox, allMail, other} {
|
||||
store.AddEntry(ctx, e)
|
||||
}
|
||||
check, _ := recordingCheck(true)
|
||||
tell, headlines, _ := recordingTell("telegram")
|
||||
if err := SendSignal(ctx, store, check, tell, signalNow); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(*headlines) != 1 || (*headlines)[0] != "3 mails matter" {
|
||||
t.Fatalf("1926/AC3: tell headlines = %v, want \"3 mails matter\"", *headlines)
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user