diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index dc0a6ad..323b673 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -8,6 +8,8 @@ on: jobs: go-test: + name: go vet + staticcheck + coverage + timeout-minutes: 25 runs-on: ubuntu-latest steps: - name: Check out @@ -53,6 +55,8 @@ jobs: awk -v total="$total" 'BEGIN { if (total + 0 < 80.0) exit 1 }' go-test-postgres: + name: postgres-backed store tests + timeout-minutes: 25 runs-on: ubuntu-latest services: postgres: @@ -86,6 +90,8 @@ jobs: run: go test -race ./internal/store go-race: + name: go test -race + timeout-minutes: 30 runs-on: ubuntu-latest steps: - name: Check out @@ -108,6 +114,8 @@ jobs: run: go test -race ./... console-build: + name: operator console build + timeout-minutes: 15 runs-on: ubuntu-latest steps: - name: Check out @@ -136,6 +144,8 @@ jobs: run: git diff --exit-code web/dist relay-test: + name: cloudflare relay tests + timeout-minutes: 15 runs-on: ubuntu-latest steps: - name: Check out @@ -156,7 +166,20 @@ jobs: working-directory: relay run: npm test + # Content-based, and authoritative. The Go suite used to infer staleness + # from mtimes, which a git checkout does not preserve — the committed + # bundle could look older than its source purely by checkout write order. + # This job already has Node, so it can just rebuild and compare bytes. + - name: Verify the committed relay bundle matches its source + working-directory: relay + run: | + npm run bundle + git diff --exit-code -- dist \ + || { echo "::error::relay/dist is stale — run 'npm run bundle' in relay/ and commit the result"; exit 1; } + docs-build: + name: docs site build + timeout-minutes: 15 runs-on: ubuntu-latest steps: - name: Check out diff --git a/.github/workflows/deploy-website.yml b/.github/workflows/deploy-website.yml index b457ab0..0fa35f4 100644 --- a/.github/workflows/deploy-website.yml +++ b/.github/workflows/deploy-website.yml @@ -20,6 +20,8 @@ concurrency: jobs: deploy: + name: deploy moltnet.dev + timeout-minutes: 15 runs-on: ubuntu-latest environment: name: github-pages diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index 964b836..9c990fc 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -10,6 +10,8 @@ permissions: jobs: build-assets: + name: build release tarballs + timeout-minutes: 20 runs-on: ubuntu-latest strategy: fail-fast: false @@ -64,6 +66,8 @@ jobs: moltnet_${{ matrix.goos }}_${{ matrix.goarch }}.tar.gz.sha256 publish-release: + name: publish GitHub release + timeout-minutes: 20 runs-on: ubuntu-latest needs: build-assets steps: diff --git a/AGENTS.md b/AGENTS.md index 7503a8c..c689815 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -33,3 +33,23 @@ Treat it as a future standalone repository that temporarily lives inside the Spa ## Extraction Constraint If a change would make `moltnet/` harder to move into its own repository later, do not do it. + +## Branches and pull requests + +**Never commit to `main`.** Every change lands through a pull request, without +exception — including one-line fixes, CI configuration, documentation, and +version bumps. Work on a branch, push it, open the PR, and let CI run. + +Direct commits to `main` bypass the checks that catch what local runs do not. +A zero-byte receipt store, a package that ships without its native binary, and +a two-week-red pipeline all reached `main` in this ecosystem while every local +gate was green — CI found them the first time it ran over the code. + +- Branch names describe the change: `feat/…`, `fix/…`, `ci/…`, `docs/…`. +- Commit messages are conventional and single-line (`feat:`, `fix:`, `docs:`, + `ci:`, `chore:`, `refactor:`, `test:`). +- Never add co-author lines, sign-offs, or AI attributions. +- Commit as you go rather than in one batch at the end, so history shows how + the work progressed. +- Merge with a merge commit rather than a squash when the individual commits + carry meaning; squashing collapses that history irreversibly. diff --git a/internal/bridge/daimon/async.go b/internal/bridge/daimon/async.go index 11326c4..868c0eb 100644 --- a/internal/bridge/daimon/async.go +++ b/internal/bridge/daimon/async.go @@ -24,10 +24,27 @@ func (c *Codec) StartControlAsync(ctx context.Context, client *loop.MoltnetClien return err } c.receipts = newReceiptTracker(store, c.token, client, config) - go c.receipts.Run(ctx) + done := make(chan struct{}) + c.asyncDone = done + go func() { + defer close(done) + c.receipts.Run(ctx) + }() return nil } +// WaitControlAsync blocks until the follower has returned. Every blocking call +// it makes is bound to the context StartControlAsync was given +// (`http.NewRequestWithContext` in fetch, and the same context through publish), +// so once the loop cancels it this returns promptly rather than waiting out a +// poll interval or an in-flight request. +func (c *Codec) WaitControlAsync() { + if c.asyncDone == nil { + return + } + <-c.asyncDone +} + func (c *Codec) ControlAccepted(config bridgeconfig.Config, event protocol.Event, acceptance loop.ControlAcceptance) error { if c.receipts == nil { return fmt.Errorf("daimon receipt follower is not running") diff --git a/internal/bridge/daimon/codec.go b/internal/bridge/daimon/codec.go index 2c6be1c..ca2ce7b 100644 --- a/internal/bridge/daimon/codec.go +++ b/internal/bridge/daimon/codec.go @@ -36,6 +36,8 @@ type Codec struct { token protocol.SecretString receiptStorePath string receipts *receiptTracker + // Closed when the follower goroutine StartControlAsync spawned returns. + asyncDone chan struct{} } type wakeRequest struct { diff --git a/internal/bridge/loop/AGENTS.md b/internal/bridge/loop/AGENTS.md index 1997c40..84080a6 100644 --- a/internal/bridge/loop/AGENTS.md +++ b/internal/bridge/loop/AGENTS.md @@ -19,3 +19,9 @@ This package holds the generic Moltnet event loop shared by runtime adapters. - keep this package runtime-agnostic - keep wire contracts explicit and small - avoid introducing dependencies back into `core/` or runtime adapters +- `RunControlLoopWithCodec` returning must mean everything it started has + stopped. A codec implementing `AsyncControlCodec` gets its context cancelled + and is then awaited through `WaitControlAsync` before the loop returns. + Cancellation is not completion: the follower it spawns can be mid-write to + durable state, so returning without the wait leaves a caller no safe point to + tear that state down. Any future async codec must honour the same barrier. diff --git a/internal/bridge/loop/control.go b/internal/bridge/loop/control.go index 6d3c956..017bf8a 100644 --- a/internal/bridge/loop/control.go +++ b/internal/bridge/loop/control.go @@ -32,6 +32,13 @@ func RunControlLoopWithCodec(ctx context.Context, config bridgeconfig.Config, co if err := asyncCodec.StartControlAsync(loopCtx, client, config); err != nil { return err } + // Cancel BEFORE waiting, and register this after `defer cancelLoop()` + // so LIFO ordering runs it first: waiting on a follower whose context + // is still live would hang instead of shutting down. + defer func() { + cancelLoop() + asyncCodec.WaitControlAsync() + }() } controlClient := &http.Client{Timeout: controlRequestTimeout} backoff := bridgeutil.NewBackoff(bridgeutil.DefaultReconnectBaseDelay, bridgeutil.DefaultReconnectMaxDelay) diff --git a/internal/bridge/loop/control_async_shutdown_test.go b/internal/bridge/loop/control_async_shutdown_test.go new file mode 100644 index 0000000..da662e9 --- /dev/null +++ b/internal/bridge/loop/control_async_shutdown_test.go @@ -0,0 +1,99 @@ +package loop + +import ( + "context" + "net/http" + "net/http/httptest" + "sync/atomic" + "testing" + "time" + + "github.com/gorilla/websocket" + "github.com/noopolis/moltnet/pkg/bridgeconfig" + "github.com/noopolis/moltnet/pkg/protocol" +) + +// asyncShutdownCodec models what the Daimon codec does: StartControlAsync +// spawns a follower bound to the loop's context, and that follower still has +// durable work to finish after cancellation (writing its receipt store). +type asyncShutdownCodec struct { + legacyControlCodec + done chan struct{} + finished atomic.Bool +} + +func (c *asyncShutdownCodec) StartControlAsync(ctx context.Context, _ *MoltnetClient, _ bridgeconfig.Config) error { + c.done = make(chan struct{}) + go func() { + defer close(c.done) + <-ctx.Done() + // The window the flake lived in: cancellation is not completion, and a + // follower that is mid-write keeps writing after the loop is told to stop. + time.Sleep(20 * time.Millisecond) + c.finished.Store(true) + }() + return nil +} + +func (c *asyncShutdownCodec) ControlAccepted(bridgeconfig.Config, protocol.Event, ControlAcceptance) error { + return nil +} + +func (c *asyncShutdownCodec) WaitControlAsync() { + if c.done == nil { + return + } + <-c.done +} + +// RunControlLoopWithCodec returning must mean every goroutine it started has +// returned. Without the barrier the loop cancels the follower and returns +// immediately, so a caller has no safe point to tear down the state that +// follower is still writing — which is how `t.TempDir()` cleanup started +// racing the Daimon receipt store. +func TestRunControlLoopWaitsForAsyncCodecBeforeReturning(t *testing.T) { + t.Parallel() + + upgrader := websocket.Upgrader{CheckOrigin: func(*http.Request) bool { return true }} + moltnetServer := httptest.NewServer(http.HandlerFunc(func(response http.ResponseWriter, request *http.Request) { + if request.URL.Path != "/v1/attach" { + response.WriteHeader(http.StatusNotFound) + return + } + connection, err := upgrader.Upgrade(response, request, nil) + if err != nil { + t.Errorf("upgrade websocket: %v", err) + return + } + defer connection.Close() + writeAttachmentHandshake(t, connection, "researcher") + _ = connection.WriteControl( + websocket.CloseMessage, + websocket.FormatCloseMessage(websocket.CloseNormalClosure, "done"), + time.Now().Add(time.Second), + ) + })) + defer moltnetServer.Close() + + config := bridgeconfig.Config{ + Agent: bridgeconfig.AgentConfig{ID: "researcher", Name: "Researcher"}, + Moltnet: bridgeconfig.MoltnetConfig{BaseURL: moltnetServer.URL, NetworkID: "local"}, + Runtime: bridgeconfig.RuntimeConfig{Kind: bridgeconfig.RuntimePi, ControlURL: moltnetServer.URL}, + Rooms: []bridgeconfig.RoomBinding{{ID: "research", Wake: bridgeconfig.WakeMentions}}, + } + + codec := &asyncShutdownCodec{} + ctx, cancel := context.WithCancel(context.Background()) + go func() { + time.Sleep(50 * time.Millisecond) + cancel() + }() + + if err := RunControlLoopWithCodec(ctx, config, codec); err != nil { + t.Fatalf("RunControlLoopWithCodec() error = %v", err) + } + + if !codec.finished.Load() { + t.Fatal("RunControlLoopWithCodec returned while its async codec was still working") + } +} diff --git a/internal/bridge/loop/control_codec.go b/internal/bridge/loop/control_codec.go index 4d6499d..a0aa7fd 100644 --- a/internal/bridge/loop/control_codec.go +++ b/internal/bridge/loop/control_codec.go @@ -68,6 +68,18 @@ type DurableAcceptanceCodec interface { type AsyncControlCodec interface { StartControlAsync(context.Context, *MoltnetClient, bridgeconfig.Config) error ControlAccepted(bridgeconfig.Config, protocol.Event, ControlAcceptance) error + // WaitControlAsync blocks until everything StartControlAsync spawned has + // returned. RunControlLoopWithCodec cancels the context it handed to + // StartControlAsync before calling this, so the wait is bounded by that + // cancellation rather than by the work itself. + // + // Without it the loop cancelled the follower and returned immediately, so + // RunControlLoopWithCodec returning did not mean the codec had stopped: a + // follower could still be publishing a reply and writing its durable + // receipt store after its caller believed shutdown was complete. A caller + // then has no safe point to tear down the receipt store's directory, and + // the goroutine outlives the loop that owns it. + WaitControlAsync() } // ControlPublicationObserver is an optional codec hook called only after a diff --git a/internal/machine/session_input_test.go b/internal/machine/session_input_test.go index 1b2ae6d..969da41 100644 --- a/internal/machine/session_input_test.go +++ b/internal/machine/session_input_test.go @@ -8,7 +8,6 @@ import ( "strings" "sync/atomic" "testing" - "time" "github.com/noopolis/moltnet/pkg/protocol" ) @@ -62,7 +61,7 @@ func TestSessionParentCancelClosesInput(t *testing.T) { select { case <-reader.closed: - case <-time.After(time.Second): + case <-waitTimeout(t): t.Fatal("expected blocking reader to be closed on parent cancel") } } diff --git a/internal/machine/session_lifetime_test.go b/internal/machine/session_lifetime_test.go index cf4e7ee..4ba18ce 100644 --- a/internal/machine/session_lifetime_test.go +++ b/internal/machine/session_lifetime_test.go @@ -6,7 +6,6 @@ import ( "io" "strings" "testing" - "time" "github.com/noopolis/moltnet/pkg/protocol" ) @@ -34,7 +33,7 @@ func TestSessionActiveCapacity(t *testing.T) { if response.Error == nil || response.Error.Code != protocol.MachineErrorCapacity { t.Fatalf("expected active-capacity rejection, got %#v", response) } - case <-time.After(2 * time.Second): + case <-waitTimeout(t): t.Fatal("timed out waiting for active-capacity rejection") } close(gate) diff --git a/internal/machine/session_test_helpers_test.go b/internal/machine/session_test_helpers_test.go index 5488423..70b8cef 100644 --- a/internal/machine/session_test_helpers_test.go +++ b/internal/machine/session_test_helpers_test.go @@ -36,11 +36,40 @@ func decodeResponses(t *testing.T, raw []byte) []protocol.MachineResponse { return responses } +// waitTimeout bounds a test's wait for a condition it expects to become true. +// +// These waits are COMPLETION waits, not rate assertions: nothing here asserts +// that the session is fast, only that it finishes. A fixed short deadline +// therefore does not test the product, it tests the hardware — and +// TestSessionActiveCapacity gates protocol.MachineMaxActiveRequests (512) +// goroutines, so under -race on a 2-core CI runner the legitimate drain +// exceeded the old fixed 2s and failed there while taking 0.5s here. +// +// The deadline is derived from the test binary's own -timeout so it scales with +// how long the run is already allowed to take, capped so a genuine deadlock +// still reports in a minute instead of stalling until the binary is killed. +func waitTimeout(t *testing.T) <-chan time.Time { + t.Helper() + const generous = 60 * time.Second + timeout := generous + if deadline, ok := t.Deadline(); ok { + // Leave headroom so this reports its own message rather than losing to + // the harness timeout, which would kill the binary instead. + if remaining := time.Until(deadline) - time.Second; remaining < timeout { + timeout = remaining + } + } + if timeout <= 0 { + timeout = time.Millisecond + } + return time.After(timeout) +} + func readLine(t *testing.T, ch <-chan string) { t.Helper() select { case <-ch: - case <-time.After(2 * time.Second): + case <-waitTimeout(t): t.Fatal("timed out waiting for executor") } } @@ -50,7 +79,7 @@ func readErr(t *testing.T, ch <-chan error) error { select { case value := <-ch: return value - case <-time.After(2 * time.Second): + case <-waitTimeout(t): t.Fatal("timed out") } return nil diff --git a/internal/relaydeploy/relay_bundle_freshness_test.go b/internal/relaydeploy/relay_bundle_freshness_test.go deleted file mode 100644 index d547f25..0000000 --- a/internal/relaydeploy/relay_bundle_freshness_test.go +++ /dev/null @@ -1,175 +0,0 @@ -package relaydeploy - -import ( - "fmt" - "io/fs" - "os" - "path/filepath" - "runtime" - "strings" - "testing" - "time" -) - -// ensureEmbeddedRelayBundleFresh mirrors internal/transport's -// ensureEmbeddedWebBundleFresh for relay/dist: it fails when any relay -// Worker source is newer than the committed bundle, so a stale -// `npm run bundle` in relay/ is caught in CI without needing Node. -func ensureEmbeddedRelayBundleFresh(sourceRoot, outputRoot string) error { - var sources []string - if err := filepath.WalkDir(filepath.Join(sourceRoot, "src"), func(path string, entry fs.DirEntry, walkErr error) error { - if walkErr != nil { - return walkErr - } - if entry.IsDir() { - return nil - } - sources = append(sources, path) - return nil - }); err != nil { - return fmt.Errorf("scan relay sources: %w", err) - } - for _, name := range []string{"package.json", "wrangler.jsonc", "build.mjs"} { - sources = append(sources, filepath.Join(sourceRoot, name)) - } - if len(sources) == 0 { - return fmt.Errorf("relay bundle freshness scan found zero source files") - } - - var outputs []string - if err := filepath.WalkDir(outputRoot, func(path string, entry fs.DirEntry, walkErr error) error { - if walkErr != nil { - return walkErr - } - if !entry.IsDir() { - outputs = append(outputs, path) - } - return nil - }); err != nil { - return fmt.Errorf("scan relay bundle: %w", err) - } - if len(outputs) == 0 { - return fmt.Errorf("relay bundle freshness scan found zero output files") - } - - oldestOutput := outputs[0] - oldestOutputInfo, err := os.Stat(oldestOutput) - if err != nil { - return fmt.Errorf("stat relay bundle output %s: %w", oldestOutput, err) - } - for _, output := range outputs[1:] { - info, statErr := os.Stat(output) - if statErr != nil { - return fmt.Errorf("stat relay bundle output %s: %w", output, statErr) - } - if info.ModTime().Before(oldestOutputInfo.ModTime()) { - oldestOutput, oldestOutputInfo = output, info - } - } - - for _, source := range sources { - info, statErr := os.Stat(source) - if statErr != nil { - return fmt.Errorf("stat relay source %s: %w", source, statErr) - } - // Git checkouts can stamp adjacent source and generated files a few - // milliseconds apart even when the committed bundle is current. Keep - // a small filesystem-timestamp tolerance, same as the web bundle - // freshness check. - if info.ModTime().After(oldestOutputInfo.ModTime().Add(2 * time.Second)) { - return fmt.Errorf( - "relay bundle stale: source %s mtime %s is newer than oldest output %s mtime %s; run npm run bundle in relay/", - source, info.ModTime().Format(time.RFC3339Nano), oldestOutput, oldestOutputInfo.ModTime().Format(time.RFC3339Nano), - ) - } - } - return nil -} - -// embeddedRelayRoot returns relay/ and relay/dist relative to this test -// file, so the check runs against the real committed bundle regardless of -// the working directory `go test` is invoked from. -func embeddedRelayRoot() (string, string) { - _, filename, _, _ := runtime.Caller(0) - root := filepath.Join(filepath.Dir(filename), "..", "..", "relay") - return root, filepath.Join(root, "dist") -} - -// TestEmbeddedRelayBundleIsFresh guards the actual committed relay/dist -// against drifting from relay/src, wrangler.jsonc, and build.mjs. -func TestEmbeddedRelayBundleIsFresh(t *testing.T) { - t.Parallel() - - relayRoot, relayDist := embeddedRelayRoot() - if err := ensureEmbeddedRelayBundleFresh(relayRoot, relayDist); err != nil { - t.Fatal(err) - } -} - -func TestEmbeddedRelayBundleFreshnessRejectsEmptyOutput(t *testing.T) { - t.Parallel() - root := t.TempDir() - if err := os.Mkdir(filepath.Join(root, "src"), 0o755); err != nil { - t.Fatal(err) - } - if err := os.WriteFile(filepath.Join(root, "src", "server.ts"), []byte("source"), 0o644); err != nil { - t.Fatal(err) - } - if err := os.Mkdir(filepath.Join(root, "dist"), 0o755); err != nil { - t.Fatal(err) - } - for _, name := range []string{"package.json", "wrangler.jsonc", "build.mjs"} { - if err := os.WriteFile(filepath.Join(root, name), []byte("{}"), 0o644); err != nil { - t.Fatal(err) - } - } - - err := ensureEmbeddedRelayBundleFresh(root, filepath.Join(root, "dist")) - if err == nil || !strings.Contains(err.Error(), "zero output files") { - t.Fatalf("expected empty-output vacuity failure, got %v", err) - } -} - -func TestEmbeddedRelayBundleFreshnessAllowsCheckoutTimestampSkew(t *testing.T) { - t.Parallel() - root := t.TempDir() - source := filepath.Join(root, "src", "server.ts") - output := filepath.Join(root, "dist", "worker.js") - for _, directory := range []string{filepath.Dir(source), filepath.Dir(output)} { - if err := os.Mkdir(directory, 0o755); err != nil { - t.Fatal(err) - } - } - for _, path := range []string{source, output} { - if err := os.WriteFile(path, []byte("fixture"), 0o644); err != nil { - t.Fatal(err) - } - } - for _, name := range []string{"package.json", "wrangler.jsonc", "build.mjs"} { - path := filepath.Join(root, name) - if err := os.WriteFile(path, []byte("fixture"), 0o644); err != nil { - t.Fatal(err) - } - } - - checkoutTime := time.Unix(1_700_000_000, 0) - if err := os.Chtimes(output, checkoutTime, checkoutTime); err != nil { - t.Fatal(err) - } - for _, name := range []string{"src/server.ts", "package.json", "wrangler.jsonc", "build.mjs"} { - path := filepath.Join(root, name) - if err := os.Chtimes(path, checkoutTime.Add(time.Second), checkoutTime.Add(time.Second)); err != nil { - t.Fatal(err) - } - } - if err := ensureEmbeddedRelayBundleFresh(root, filepath.Join(root, "dist")); err != nil { - t.Fatalf("expected checkout timestamp skew to pass: %v", err) - } - - if err := os.Chtimes(source, checkoutTime.Add(3*time.Second), checkoutTime.Add(3*time.Second)); err != nil { - t.Fatal(err) - } - if err := ensureEmbeddedRelayBundleFresh(root, filepath.Join(root, "dist")); err == nil || !strings.Contains(err.Error(), "relay bundle stale") { - t.Fatalf("expected genuinely stale source to fail, got %v", err) - } -}