From 71413a4248c3da303e962ce817c3da3474219681 Mon Sep 17 00:00:00 2001 From: zhangwenjian Date: Sat, 5 Sep 2026 22:31:11 +0800 Subject: [PATCH 1/9] =?UTF-8?q?build=F0=9F=94=A7:=20require=20go-admin-cor?= =?UTF-8?q?e=20v2.6.0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit It carries the life-cycle phases and the shutdown registry. Nothing here uses them yet - the wiring is the commits that follow, and keeping the bump on its own means a bisect can tell "the dependency moved" apart from "the host started calling into it". --- go.mod | 2 +- go.sum | 2 ++ 2 files changed, 3 insertions(+), 1 deletion(-) diff --git a/go.mod b/go.mod index 8f993281..899f5147 100644 --- a/go.mod +++ b/go.mod @@ -11,7 +11,7 @@ require ( github.com/casbin/casbin/v3 v3.8.1 github.com/gin-gonic/gin v1.12.0 github.com/glebarez/sqlite v1.11.0 - github.com/go-admin-team/go-admin-core/v2 v2.5.0 + github.com/go-admin-team/go-admin-core/v2 v2.6.0 github.com/google/uuid v1.6.0 github.com/huaweicloud/huaweicloud-sdk-go-obs v3.26.6+incompatible github.com/mssola/user_agent v0.6.0 diff --git a/go.sum b/go.sum index fbfb007f..a79c5d22 100644 --- a/go.sum +++ b/go.sum @@ -147,6 +147,8 @@ github.com/glebarez/sqlite v1.11.0 h1:wSG0irqzP6VurnMEpFGer5Li19RpIRi2qvQz++w0GM github.com/glebarez/sqlite v1.11.0/go.mod h1:h8/o8j5wiAsqSPoWELDUdJXhjAhsVliSn7bWZjOhrgQ= github.com/go-admin-team/go-admin-core/v2 v2.5.0 h1:aD1SALklBxizGB9u8cOgm4OT8z656FM83F4fD6dMz9g= github.com/go-admin-team/go-admin-core/v2 v2.5.0/go.mod h1:LG/XvEfOplbuadKrPTPm0Nu5pN06aQUNZZC3ao4B4gs= +github.com/go-admin-team/go-admin-core/v2 v2.6.0 h1:sRoZaxniTpbe287uR/uWpA14Jl1GTAcfGXLKoBLph2w= +github.com/go-admin-team/go-admin-core/v2 v2.6.0/go.mod h1:LG/XvEfOplbuadKrPTPm0Nu5pN06aQUNZZC3ao4B4gs= github.com/go-kit/kit v0.8.0/go.mod h1:xBxKIO96dXMWWy0MnWVtmwkA9/13aqxPnvrjFYMA2as= github.com/go-kit/kit v0.9.0/go.mod h1:xBxKIO96dXMWWy0MnWVtmwkA9/13aqxPnvrjFYMA2as= github.com/go-kit/kit v0.10.0/go.mod h1:xUsJbQ/Fp4kEt7AFgCuvyX4a71u8h9jB8tj/ORgOZ7o= From d52dca1cb685845d0cb6d42a0cdc2c17196791e3 Mon Sep 17 00:00:00 2001 From: zhangwenjian Date: Sat, 5 Sep 2026 22:31:11 +0800 Subject: [PATCH 2/9] =?UTF-8?q?feat=E2=9C=A8:=20announce=20BeforeRouter=20?= =?UTF-8?q?and=20AfterListen,=20and=20bind=20before=20either?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two phases are now announced from the command that serves traffic, so a module can attach to them instead of being called by name from here. The listener is opened by this goroutine rather than left to ListenAndServe, which binds on the goroutine that serves. That mattered for the phase: a hook on AfterListen is promised a reachable port, and with the bind happening out of sight there was no way to keep that promise - "address already in use" surfaced on a goroutine nobody read, after the banner had already announced the server was up. It is now returned from run() and the process exits non-zero without claiming anything. AfterListen is announced synchronously. Moving it to a goroutine to save the few milliseconds would let it overlap the shutdown, and on a fast SIGTERM the cleanup callbacks could finish before the startup ones did. What is left in the serving goroutine is still log.Fatal, deliberately: the bind is no longer among the errors that reach it, so what remains is a serve that failed after the port was taken, and carrying on would park the process on <-quit with nothing serving. ServeTLS is the one case that can still fail immediately, since it reads the certificate files - with ssl enabled a hook can still run against a server on its way down. That is not a regression (the old code printed the banner in the same situation) and it is not fixed here. BeforeRouter is placed before initRouter, which is a different moment from the before registry runStartupHooks drains: those callbacks run after the engine has been built, not before it. AfterListen is tested here, in one test rather than two because the phase seals itself once it has run: a second test would find a closed registry and pass while proving nothing. Both counter-proofs compile and fail - announcing on a failed bind reports "AfterListen ran 1 times after a failed bind", and `go RunPhase(...)` reports "ran 0 times, want 1" against the hook's own pause. BeforeRouter's placement is not asserted in this commit. The test for it comes with the buildRouter extraction later in this branch. --- cmd/api/phase_test.go | 88 +++++++++++++++++++++++++++++++++++++++++++ cmd/api/server.go | 68 +++++++++++++++++++++++++++------ 2 files changed, 144 insertions(+), 12 deletions(-) create mode 100644 cmd/api/phase_test.go diff --git a/cmd/api/phase_test.go b/cmd/api/phase_test.go new file mode 100644 index 00000000..20d5b80c --- /dev/null +++ b/cmd/api/phase_test.go @@ -0,0 +1,88 @@ +package api + +import ( + "fmt" + "net" + "net/http" + "testing" + "time" + + "github.com/go-admin-team/go-admin-core/v2/sdk" + "github.com/go-admin-team/go-admin-core/v2/sdk/runtime" +) + +// freePort returns a port nothing is listening on. It is inherently a guess - +// the port is free when it is handed back and could be taken a moment later - +// but every alternative needs the caller to hold the listener, which is the one +// thing these tests cannot do. +func freePort(t *testing.T) int { + t.Helper() + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("probe listen: %v", err) + } + port := ln.Addr().(*net.TCPAddr).Port + _ = ln.Close() + return port +} + +// AfterListen promises a hook that the port is reachable. Both halves of that +// are asserted here, and in one test rather than two, because the phase seals +// itself once it has run: a second test calling RunPhase again would find a +// closed registry and pass while proving nothing. +// +// The failing bind comes first for the same reason. It must leave the phase +// unsealed, which is only visible if nothing has sealed it yet. +func TestAfterListenIsAnnouncedOnlyOnceThePortIsBound(t *testing.T) { + // The pause makes the "announced synchronously" claim testable: if the + // announcement were moved onto a goroutine, startServing would return + // while the hook was still sleeping and the count below would be zero. + var ran int + sdk.Runtime.SetPhase(runtime.AfterListen, func() { + time.Sleep(50 * time.Millisecond) + ran++ + }) + + // Somebody else already has the port. Under ListenAndServe this surfaced + // on the serving goroutine, far too late to stop the announcement. + taken, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("occupy: %v", err) + } + defer func() { _ = taken.Close() }() + + blocked := &http.Server{Addr: taken.Addr().String(), Handler: http.NewServeMux()} + if err := startServing(blocked, false, "", ""); err == nil { + t.Fatal("startServing returned no error for a port that was already taken") + } + if ran != 0 { + t.Errorf("AfterListen ran %d times after a failed bind; a hook there is told the port is reachable", ran) + } + if sdk.Runtime.PhaseSealed(runtime.AfterListen) { + t.Error("a failed bind sealed AfterListen, so the phase could never run for a server that did start") + } + + // And now a bind that works. + port := freePort(t) + srv := &http.Server{Addr: fmt.Sprintf("127.0.0.1:%d", port), Handler: http.NewServeMux()} + if err := startServing(srv, false, "", ""); err != nil { + t.Fatalf("startServing on a free port: %v", err) + } + defer func() { _ = srv.Close() }() + + // Checked the instant startServing returns, so this is also the assertion + // that it did not return early: an asynchronous announcement would still + // be inside the sleep. Synchrony matters because an announcement that + // overlaps the wait below could, on a fast SIGTERM, have the shutdown + // callbacks finish before the startup ones. + if ran != 1 { + t.Fatalf("AfterListen ran %d times, want 1", ran) + } + + // The claim is not "Serve was called" but "the port answers". Dial it. + c, err := net.DialTimeout("tcp", srv.Addr, 5*time.Second) + if err != nil { + t.Fatalf("AfterListen ran but the port does not answer: %v", err) + } + _ = c.Close() +} diff --git a/cmd/api/server.go b/cmd/api/server.go index 646b988b..cdb0cad1 100644 --- a/cmd/api/server.go +++ b/cmd/api/server.go @@ -3,6 +3,7 @@ package api import ( "context" "fmt" + "net" "net/http" "os" "os/signal" @@ -16,6 +17,7 @@ import ( "github.com/go-admin-team/go-admin-core/v2/sdk/api" "github.com/go-admin-team/go-admin-core/v2/sdk/config" "github.com/go-admin-team/go-admin-core/v2/sdk/pkg" + "github.com/go-admin-team/go-admin-core/v2/sdk/runtime" "github.com/pkg/errors" "github.com/spf13/cobra" @@ -81,6 +83,12 @@ func run() error { if config.ApplicationConfig.Mode == pkg.ModeProd.String() { gin.SetMode(gin.ReleaseMode) } + // The last point at which a module can still affect how routes are built. + // It is not the same moment as the before registry, which runStartupHooks + // drains below - those callbacks run after initRouter has built the engine, + // not before it. + sdk.Runtime.RunPhase(runtime.BeforeRouter) + initRouter() runStartupHooks() @@ -122,18 +130,10 @@ func run() error { // arming is separate from waiting. quit, disarmStopSignals := armStopSignals() - go func() { - // 服务连接 - if config.SslConfig.Enable { - if err := srv.ListenAndServeTLS(config.SslConfig.Pem, config.SslConfig.KeyStr); err != nil && !errors.Is(err, http.ErrServerClosed) { - log.Fatal("listen: ", err) - } - } else { - if err := srv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) { - log.Fatal("listen: ", err) - } - } - }() + if err := startServing(srv, config.SslConfig.Enable, config.SslConfig.Pem, config.SslConfig.KeyStr); err != nil { + return err + } + fmt.Println(pkg.Red(string(global.LogoContent))) tip() fmt.Println(pkg.Green("Server run at:")) @@ -185,6 +185,50 @@ func armStopSignals() (<-chan os.Signal, func()) { return quit, func() { signal.Stop(quit) } } +// startServing binds srv.Addr, hands the listener to srv on its own goroutine, +// and announces AfterListen. +// +// The bind is done here rather than left to ListenAndServe, which binds on the +// goroutine that serves. That put the failure every deployment actually hits - +// "address already in use" - on a goroutine nobody was reading, so the banner +// went on to claim the server was up, and there would be no way to keep +// AfterListen from announcing a socket that does not exist. A hook there is +// promised a reachable port; the only way to keep that promise is for the bind +// to have already happened on this goroutine. +// +// AfterListen is announced synchronously. Running it in a goroutine to save the +// few milliseconds would let it overlap the shutdown: on a fast SIGTERM the +// cleanup callbacks could finish before the startup ones had. +func startServing(srv *http.Server, useTLS bool, pem, key string) error { + ln, err := net.Listen("tcp", srv.Addr) + if err != nil { + return errors.Wrap(err, "listen") + } + + go func() { + // 服务连接 + var err error + if useTLS { + err = srv.ServeTLS(ln, pem, key) + } else { + err = srv.Serve(ln) + } + if err != nil && !errors.Is(err, http.ErrServerClosed) { + // Still fatal, as it was. The bind is no longer among the errors + // that reach here; what is left is a serve that failed after the + // port was taken, and carrying on would park the process on + // <-quit with nothing serving. TLS is the one case that can still + // fail immediately - ServeTLS reads the certificate files - so + // with ssl enabled AfterListen can run against a server that is + // already on its way down. + log.Fatal("listen: ", err) + } + }() + + sdk.Runtime.RunPhase(runtime.AfterListen) + return nil +} + // shutdownServer stops srv, giving in-flight requests up to timeout to finish. // // It returns the error instead of exiting on it. A caller that exits here skips From 750c7c744ec1273ed7e6f22fadbc8a8b517b75aa Mon Sep 17 00:00:00 2001 From: zhangwenjian Date: Sat, 5 Sep 2026 22:31:12 +0800 Subject: [PATCH 3/9] =?UTF-8?q?feat=E2=9C=A8:=20run=20the=20BeforeExit=20c?= =?UTF-8?q?allbacks=20on=20the=20way=20out?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A module can now register cleanup and have it happen. Until this commit the process stopped serving and returned; anything a module had set up went down with the process rather than being taken down. BeginShutdown is said first, before anything is dismantled. Without it a configuration reload arriving in this window re-runs AfterResource - rebuilding the pool and the queue adapter and re-registering consumers - on top of cleanup that has already run. The cleanup runs whether or not Shutdown reported an error, which is the whole reason that error stopped being fatal in the first place: Shutdown fails exactly when connections were still in flight, and that is when there is most left to take down. The two budgets are spent one after the other, so what has to fit inside the orchestrator's grace period is their sum. `docker stop` allows 10s by default before SIGKILL; 5+3 leaves room to finish returning. Raising one without lowering the other buys nothing. Both halves are tested through the existing subprocess child, which now registers a BeforeExit callback of its own: - after a Shutdown that timed out, the callback still runs. Moving the call into the success branch reports "the BeforeExit callback did not run after a failed Shutdown". - a callback that outlasts its budget is abandoned, not awaited. It sleeps two seconds against a 300ms budget; RunShutdown reports the deadline, the process exits cleanly inside one second, and the callback's own marker never appears. Widening the budget to five seconds makes the test time out waiting for the exit, which is what "awaited" looks like. Both counter-proofs compile and fail. --- cmd/api/server.go | 40 +++++++++++++++++-- cmd/api/signal_test.go | 91 ++++++++++++++++++++++++++++++++++++++---- 2 files changed, 120 insertions(+), 11 deletions(-) diff --git a/cmd/api/server.go b/cmd/api/server.go index cdb0cad1..1decd059 100644 --- a/cmd/api/server.go +++ b/cmd/api/server.go @@ -149,6 +149,12 @@ func run() error { // the default handler, so a shutdown that hangs can still be interrupted. disarmStopSignals() + // Said before anything is taken apart. A configuration reload arriving in + // this window would otherwise re-run AfterResource - rebuilding the pool + // and the queue adapter, and re-registering consumers - on top of cleanup + // that has already run. + sdk.Runtime.BeginShutdown() + log.Info("Shutdown Server ... ") if err := shutdownServer(srv, shutdownTimeout); err != nil { // Not log.Fatal: that is an unconditional os.Exit(1), and Shutdown @@ -156,15 +162,28 @@ func run() error { // which is when the cleanup that follows matters most. log.Error("Server Shutdown: ", err) } + + // Runs whether or not the line above reported an error, for that reason. + if err := runShutdownHooks(cleanupTimeout); err != nil { + log.Error("Cleanup: ", err) + } log.Info("Server exiting") return nil } -// shutdownTimeout is how long Shutdown waits for in-flight requests. It plus -// whatever cleanup follows has to stay inside the orchestrator's grace period -// - `docker stop` allows 10s by default before it sends SIGKILL. -const shutdownTimeout = 5 * time.Second +// shutdownTimeout is how long Shutdown waits for in-flight requests, and +// cleanupTimeout how long the BeforeExit callbacks get after it. +// +// They are consumed one after the other, so the two together are what has to +// stay inside the orchestrator's grace period: `docker stop` allows 10s by +// default before it sends SIGKILL, and 5+3 leaves room for the process to +// finish returning. Raising either without lowering the other buys nothing - +// the budget that runs out is the orchestrator's. +const ( + shutdownTimeout = 5 * time.Second + cleanupTimeout = 3 * time.Second +) // armStopSignals registers for the stop signals and returns the channel they // arrive on together with the function that restores the default disposition. @@ -240,6 +259,19 @@ func shutdownServer(srv *http.Server, timeout time.Duration) error { return srv.Shutdown(ctx) } +// runShutdownHooks runs the BeforeExit callbacks with timeout to share. +// +// What the budget bounds is the wait, not the work. When it is gone RunShutdown +// stops waiting and returns; a callback that never looks at its context carries +// on until the process exits, and may leave a partial write behind. Go cannot +// cancel a function that does not check for cancellation, which is why the +// callbacks are handed a context at all. +func runShutdownHooks(timeout time.Duration) error { + ctx, cancel := context.WithTimeout(context.Background(), timeout) + defer cancel() + return sdk.Runtime.RunShutdown(ctx) +} + // runStartupHooks runs the router registries and then the before callbacks. // // The package-level slice runs first and in its existing order, so a fork that diff --git a/cmd/api/signal_test.go b/cmd/api/signal_test.go index 3e3c5531..7d296d6f 100644 --- a/cmd/api/signal_test.go +++ b/cmd/api/signal_test.go @@ -1,6 +1,7 @@ package api import ( + "context" "fmt" "net" "net/http" @@ -10,6 +11,8 @@ import ( "syscall" "testing" "time" + + "github.com/go-admin-team/go-admin-core/v2/sdk" ) // The signal path cannot be exercised in-process: delivering a signal to the @@ -21,13 +24,15 @@ import ( // this repository's CI has no database (.github/workflows/go.yml runs neither // MySQL nor a sqlite-tagged build), and none of what is under test needs one. const ( - childEnv = "GO_ADMIN_SIGNAL_CHILD" - childStuckEnv = "GO_ADMIN_SIGNAL_CHILD_STUCK" - childHangConn = "GO_ADMIN_SIGNAL_CHILD_HANGCONN" - markerReady = "CHILD-READY" - markerSignal = "CHILD-SIGNAL" - markerShutdown = "CHILD-SHUTDOWN-OK" - markerExiting = "CHILD-EXITING" + childEnv = "GO_ADMIN_SIGNAL_CHILD" + childStuckEnv = "GO_ADMIN_SIGNAL_CHILD_STUCK" + childHangConn = "GO_ADMIN_SIGNAL_CHILD_HANGCONN" + childSlowCleanup = "GO_ADMIN_SIGNAL_CHILD_SLOWCLEANUP" + markerReady = "CHILD-READY" + markerSignal = "CHILD-SIGNAL" + markerShutdown = "CHILD-SHUTDOWN-OK" + markerCleanup = "CHILD-CLEANUP-RAN" + markerExiting = "CHILD-EXITING" ) // TestSignalChild is the child process. It is skipped in a normal run. @@ -59,6 +64,24 @@ func TestSignalChild(t *testing.T) { } go func() { _ = srv.Serve(ln) }() + // A BeforeExit callback, registered the way a module would. What the tests + // below care about is whether it runs at all - after a Shutdown that + // failed, and after its own budget has been spent. + cleanupBudget := cleanupTimeout + sdk.Runtime.SetShutdown(func(ctx context.Context) { + if os.Getenv(childSlowCleanup) == "1" { + // Outlasts the budget on purpose, and does not consult ctx - + // which is the case the contract is explicit about: what the + // context bounds is the wait, not the work. + time.Sleep(2 * time.Second) + } + fmt.Println(markerCleanup) + os.Stdout.Sync() + }) + if os.Getenv(childSlowCleanup) == "1" { + cleanupBudget = 300 * time.Millisecond + } + // Arm before announcing readiness. Doing it the other way round leaves a // window in which the parent's signal reaches the default handler and // kills the child before any of this runs - which is exactly the failure @@ -110,6 +133,8 @@ func TestSignalChild(t *testing.T) { timeout = 300 * time.Millisecond } + sdk.Runtime.BeginShutdown() + if err := shutdownServer(srv, timeout); err != nil { // Deliberately not fatal, and deliberately not a bare return: the // point is that whatever follows still runs. @@ -117,6 +142,10 @@ func TestSignalChild(t *testing.T) { } else { fmt.Println(markerShutdown) } + + if err := runShutdownHooks(cleanupBudget); err != nil { + fmt.Println("cleanup error:", err) + } fmt.Println(markerExiting) os.Stdout.Sync() } @@ -286,7 +315,55 @@ func TestShutdownTimeoutDoesNotStopWhatFollows(t *testing.T) { t.Fatalf("Shutdown did not time out, so this test proves nothing; saw:\n%s", strings.Join(seen, "\n")) } + var cleaned bool + for _, l := range seen { + if strings.Contains(l, markerCleanup) { + cleaned = true + } + } + if !cleaned { + t.Fatalf("the BeforeExit callback did not run after a failed Shutdown; saw:\n%s", + strings.Join(seen, "\n")) + } if err := cmd.Wait(); err != nil { t.Fatalf("child exited with %v after a failed Shutdown, want a clean exit", err) } } + +// A callback that outlasts its budget must not take the process with it, and +// must not be waited for: RunShutdown reports the deadline and returns, the +// callback carries on, and the process still exits cleanly. This is the half of +// the contract that is easy to get backwards - the context bounds the wait, not +// the work, because Go cannot cancel a function that does not check for it. +func TestACleanupThatOutlastsItsBudgetIsAbandonedNotAwaited(t *testing.T) { + cmd, _, lines := startChild(t, false, childSlowCleanup+"=1") + await(t, lines, markerReady, 30*time.Second) + + if err := cmd.Process.Signal(syscall.SIGTERM); err != nil { + t.Fatalf("signal: %v", err) + } + await(t, lines, markerSignal, 10*time.Second) + + // The budget is 300ms and the callback sleeps two seconds. If RunShutdown + // waited for it, this marker would not arrive for two seconds; the one + // second here is what makes "abandoned, not awaited" the thing asserted. + seen := await(t, lines, markerExiting, 1*time.Second) + + var reported bool + for _, l := range seen { + if strings.Contains(l, "cleanup error:") { + reported = true + } + if strings.Contains(l, markerCleanup) { + t.Fatalf("the slow callback finished before the process moved on, so nothing was abandoned; saw:\n%s", + strings.Join(seen, "\n")) + } + } + if !reported { + t.Fatalf("RunShutdown returned no error for a callback that outlasted the budget; saw:\n%s", + strings.Join(seen, "\n")) + } + if err := cmd.Wait(); err != nil { + t.Fatalf("child exited with %v, want a clean exit despite the abandoned callback", err) + } +} From 4510b06959da3ddeb8f0fc6b5964616f810e159a Mon Sep 17 00:00:00 2001 From: zhangwenjian Date: Sat, 5 Sep 2026 22:31:12 +0800 Subject: [PATCH 4/9] =?UTF-8?q?fix=F0=9F=90=9B:=20register=20the=20queue?= =?UTF-8?q?=20consumers=20before=20the=20queue=20is=20started?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit setupQueue ended with `go queueAdapter.Run()`, and the three log consumers were registered afterwards, from setup(). The contract implementations refuse a registration once the queue is running - memqueue and the redis queue both answer storage.ErrQueueAlreadyStarted - and Register cannot report it: it returns nothing, which its own comment in core records as the reason the interface is deprecated. Start first and register second, across two goroutines, and the registration is dropped without the caller being able to tell. What follows is not quiet. No consumer group was created, so redis refuses every later publish with storage.ErrNoHandler, and go-admin logs that at error level from both call sites while the login and operation log rows are simply never written. The silence is in the registration; the cost shows up on every request after it. Which implementation is behind the interface depends on the configuration. config.QueueConfig.Setup returns queue.NewMemory directly when there is no redis section - and that one does not care about the order, because its Register just starts another consumer goroutine. Only a redis section reaches storage.LegacyQueueAdapter, which wraps the contract implementation and therefore refuses. So the defect is invisible in the default deployment and shows up only where redis is configured, dropping the login log, the operation log and the api check - the three things #892 was about. The start therefore moves to the code that registers, and nothing starts the queue but that. The registration also moves onto AfterResource. It has to: a reload rebuilds the adapter, and consumers attached to the one that existed at start-up are attached to a queue nobody publishes to any more. Being on that phase means running again on every reload, so the callback is idempotent with respect to a given queue rather than "does nothing the second time" - registering twice on the same queue would give every message two consumers and write every log row twice. Identity for that comes from common/storage, where the adapter is built, as a generation counter. It cannot come from the accessors: GetQueueAdapter and GetQueuePrefix build a fresh runtime.Queue wrapper on every call, so comparing two of them compares two wrappers and never matches however many times the adapter underneath has been replaced. A counter also keeps the comparison on a uint64 rather than an `==` between two interface values, which would panic on an adapter type that is not comparable. Generation 0 means the configuration has no queue section, so nothing was installed and the runtime hands back its own memory queue. That case still gets consumers, because the registration this replaces was unconditional and dropping it would stop the logs for anyone who commented the section out. Two things fixed on the way past: - `if q := sdk.Runtime.GetQueueAdapter(); q != nil { q.Shutdown() }` was always true. GetQueueAdapter never returns nil - with nothing configured the runtime falls back to its own memory queue and wraps that - so the first start shut down the fallback queue before anything had used it. Only an adapter this package installed is shut down now. - config.Setup becomes bootstrap.SetupConfig, which is what announces AfterResource, and announces it after the callbacks that build the resources rather than before. attachConsumersOnce is split out so the order and the once-ness can be checked against a queue the test controls; neither can be read back out of a real adapter. Four tests cover the ordering, both directions of the idempotency rule, and the unconfigured case. Both counter-proofs compile and fail: calling Run before the registrations reports each of the three as "came after Run", and dropping the generation guard reports eight calls where four are wanted. One honest limit: the counter-proof for the ordering makes Run synchronous. The original arrangement started the queue on another goroutine, and a race cannot be made to fail every time - which is the reason the order is enforced by structure here instead of being left to be noticed in use. --- cmd/api/queue_test.go | 167 +++++++++++++++++++++++++++++++++++ cmd/api/server.go | 71 +++++++++++++-- common/storage/initialize.go | 57 +++++++++++- 3 files changed, 284 insertions(+), 11 deletions(-) create mode 100644 cmd/api/queue_test.go diff --git a/cmd/api/queue_test.go b/cmd/api/queue_test.go new file mode 100644 index 00000000..b4b0cc86 --- /dev/null +++ b/cmd/api/queue_test.go @@ -0,0 +1,167 @@ +package api + +import ( + "strings" + "sync" + "testing" + "time" + + corestorage "github.com/go-admin-team/go-admin-core/v2/storage" +) + +// recordingQueue records what was done to it, in order. Register and Run are +// the two calls whose order is the point of this file; Append and Shutdown are +// here to satisfy the interface. +type recordingQueue struct { + mu sync.Mutex + events []string + ran chan struct{} +} + +func newRecordingQueue() *recordingQueue { + return &recordingQueue{ran: make(chan struct{}, 4)} +} + +func (q *recordingQueue) record(e string) { + q.mu.Lock() + q.events = append(q.events, e) + q.mu.Unlock() +} + +func (q *recordingQueue) seen() []string { + q.mu.Lock() + defer q.mu.Unlock() + return append([]string(nil), q.events...) +} + +func (q *recordingQueue) String() string { return "recording" } +func (q *recordingQueue) Append(corestorage.Messager) error { return nil } +func (q *recordingQueue) Register(name string, _ corestorage.ConsumerFunc) { + q.record("register:" + name) +} +func (q *recordingQueue) Shutdown() {} + +func (q *recordingQueue) Run() { + q.record("run") + select { + case q.ran <- struct{}{}: + default: + } +} + +// waitForRun waits for Run, which is started on a goroutine. +func (q *recordingQueue) waitForRun(t *testing.T) { + t.Helper() + select { + case <-q.ran: + case <-time.After(5 * time.Second): + t.Fatalf("Run was never called; saw: %v", q.seen()) + } +} + +// The consumers must be registered before the queue is started. A queue that +// is already running refuses further registration - the contract +// implementations answer storage.ErrQueueAlreadyStarted - and the legacy +// adapter this path goes through drops that error, so the wrong order loses +// consumers with nothing said about it. The memory backend does not care, +// which is exactly why this cannot be left to be noticed in use. +func TestConsumersAreRegisteredBeforeTheQueueIsStarted(t *testing.T) { + attachedQueue.Store(0) + t.Cleanup(func() { attachedQueue.Store(0) }) + + q := newRecordingQueue() + attachConsumersOnce(1, q) + q.waitForRun(t) + + seen := q.seen() + runAt := -1 + registers := 0 + for i, e := range seen { + switch { + case e == "run": + if runAt < 0 { + runAt = i + } + case strings.HasPrefix(e, "register:"): + registers++ + if runAt >= 0 { + t.Errorf("%q came after Run; a running queue refuses registration", e) + } + } + } + if registers != 3 { + t.Errorf("registered %d consumers, want 3; saw %v", registers, seen) + } + if runAt < 0 { + t.Errorf("the queue was never started; saw %v", seen) + } +} + +// AfterResource runs again on every configuration reload, so the hook has to +// be idempotent with respect to a given queue - not "does nothing the second +// time". Registering twice on the same queue would give every message two +// consumers and write every log row twice. +func TestTheSameQueueIsNotGivenConsumersTwice(t *testing.T) { + attachedQueue.Store(0) + t.Cleanup(func() { attachedQueue.Store(0) }) + + q := newRecordingQueue() + attachConsumersOnce(1, q) + q.waitForRun(t) + attachConsumersOnce(1, q) + + // Nothing to wait for on the second call, so give a wrong implementation + // the time it would need to show up. + time.Sleep(200 * time.Millisecond) + if n := len(q.seen()); n != 4 { + t.Errorf("%d calls after attaching twice to the same queue, want 4 (3 registers + 1 run); saw %v", n, q.seen()) + } +} + +// The other half of the same rule: a reload builds a new adapter, and the +// consumers on the old one are attached to a queue nobody publishes to any +// more. A new generation must get its own set. +func TestANewQueueGetsItsOwnConsumers(t *testing.T) { + attachedQueue.Store(0) + t.Cleanup(func() { attachedQueue.Store(0) }) + + first := newRecordingQueue() + attachConsumersOnce(1, first) + first.waitForRun(t) + + second := newRecordingQueue() + attachConsumersOnce(2, second) + second.waitForRun(t) + + if n := len(second.seen()); n != 4 { + t.Errorf("the queue from the second generation saw %d calls, want 4; saw %v", n, second.seen()) + } + if n := len(first.seen()); n != 4 { + t.Errorf("the queue from the first generation saw %d calls, want 4 - it should not have been touched again; saw %v", n, first.seen()) + } +} + +// Generation 0 means the configuration has no queue section at all, so nothing +// was installed and the runtime hands back its own memory queue. That case +// still has to get consumers - the registration it replaces was unconditional, +// and dropping it would stop the login and operation logs for anyone who +// commented the section out. +func TestAnUnconfiguredQueueStillGetsConsumers(t *testing.T) { + attachedQueue.Store(0) + t.Cleanup(func() { attachedQueue.Store(0) }) + + q := newRecordingQueue() + attachConsumersOnce(0, q) + q.waitForRun(t) + + if n := len(q.seen()); n != 4 { + t.Errorf("an unconfigured queue saw %d calls, want 4; saw %v", n, q.seen()) + } + + // And still only once. + attachConsumersOnce(0, q) + time.Sleep(200 * time.Millisecond) + if n := len(q.seen()); n != 4 { + t.Errorf("generation 0 was attached to twice: %d calls, want 4; saw %v", n, q.seen()) + } +} diff --git a/cmd/api/server.go b/cmd/api/server.go index 1decd059..ca510096 100644 --- a/cmd/api/server.go +++ b/cmd/api/server.go @@ -7,6 +7,7 @@ import ( "net/http" "os" "os/signal" + "sync/atomic" "syscall" "time" @@ -15,9 +16,11 @@ import ( log "github.com/go-admin-team/go-admin-core/v2/logger" "github.com/go-admin-team/go-admin-core/v2/sdk" "github.com/go-admin-team/go-admin-core/v2/sdk/api" + "github.com/go-admin-team/go-admin-core/v2/sdk/bootstrap" "github.com/go-admin-team/go-admin-core/v2/sdk/config" "github.com/go-admin-team/go-admin-core/v2/sdk/pkg" "github.com/go-admin-team/go-admin-core/v2/sdk/runtime" + corestorage "github.com/go-admin-team/go-admin-core/v2/storage" "github.com/pkg/errors" "github.com/spf13/cobra" @@ -62,23 +65,77 @@ func init() { func setup() { // 注入配置扩展项 config.ExtendConfig = &ext.ExtConfig + + // Registered before the configuration is read. SetupConfig announces + // AfterResource as soon as the callbacks that build the resources have + // run, so a hook added after that call would miss the first round and the + // queue would have no consumers until somebody edited the config file. + sdk.Runtime.SetPhase(runtime.AfterResource, attachQueueConsumers) + //1. 读取配置 - config.Setup( + bootstrap.SetupConfig( file.NewSource(file.WithPath(configYml)), database.Setup, storage.Setup, ) - //注册监听函数 - queue := sdk.Runtime.GetQueuePrefix("") - queue.Register(global.LoginLog, models.SaveLoginLog) - queue.Register(global.OperateLog, models.SaveOperaLog) - queue.Register(global.ApiCheck, models.SaveSysApi) - go queue.Run() usageStr := `starting api server...` log.Info(usageStr) } +// attachedQueue is the queue generation the consumers are attached to, plus +// one, so that the zero value means "attached to nothing yet". Written from +// the goroutine running the phase, read from the next one - rounds never +// overlap, but they are not the same goroutine. +var attachedQueue atomic.Uint64 + +// attachQueueConsumers registers the log consumers against the queue that is +// current, and starts it. +// +// It runs on AfterResource, so it runs again after every configuration reload +// - and it has to. A reload rebuilds the queue adapter, and consumers +// registered against the one that existed at start-up are attached to an +// adapter nobody publishes to any more, so the login and operation logs stop +// being written with nothing said about it. +// +// It is therefore idempotent with respect to a given queue rather than "does +// nothing the second time": a new adapter gets a fresh set of consumers, the +// same one gets none. Registering twice on the same queue would give every +// message two consumers and write every log row twice. +// +// Generation 0 means the configuration has no queue section, so nothing was +// installed and GetQueuePrefix hands back the runtime's own memory queue. +// That case still gets consumers - it is what the previous unconditional +// registration did, and dropping it would silently stop logging for anyone who +// commented the section out - it just never gets them twice. +func attachQueueConsumers() { + attachConsumersOnce(storage.QueueGeneration(), sdk.Runtime.GetQueuePrefix("")) +} + +// attachConsumersOnce puts the log consumers on q and starts it, unless gen +// says this queue already has them. +// +// Split out from attachQueueConsumers so that the order and the once-ness can +// be checked against a queue the test controls: the sequence that matters here +// cannot be read back out of a real adapter. +func attachConsumersOnce(gen uint64, q corestorage.AdapterQueue) { + if attachedQueue.Load() == gen+1 { + return + } + attachedQueue.Store(gen + 1) + + //注册监听函数 + q.Register(global.LoginLog, models.SaveLoginLog) + q.Register(global.OperateLog, models.SaveOperaLog) + q.Register(global.ApiCheck, models.SaveSysApi) + + // Started only now, and by whoever registered. setupQueue deliberately + // leaves it stopped: a queue that is already running refuses further + // registration, and the adapter in this path drops that error on the + // floor, so starting first loses consumers without a word. + go q.Run() +} + func run() error { if config.ApplicationConfig.Mode == pkg.ModeProd.String() { gin.SetMode(gin.ReleaseMode) diff --git a/common/storage/initialize.go b/common/storage/initialize.go index 9a5f563a..b1f552cb 100644 --- a/common/storage/initialize.go +++ b/common/storage/initialize.go @@ -9,10 +9,11 @@ package storage import ( "log" + "sync" + "github.com/go-admin-team/go-admin-core/v2/captcha" "github.com/go-admin-team/go-admin-core/v2/sdk" "github.com/go-admin-team/go-admin-core/v2/sdk/config" - "github.com/go-admin-team/go-admin-core/v2/captcha" ) // Setup 配置storage组件 @@ -34,17 +35,65 @@ func setupCaptcha() { captcha.SetStore(captcha.NewCacheStore(sdk.Runtime.GetCacheAdapter(), 600)) } +var ( + queueMu sync.Mutex + // installed is the adapter setupQueue built, kept so the next reload can + // shut it down, and counted so a consumer can tell one from the next. + installed interface{ Shutdown() } + installedGen uint64 +) + +// QueueGeneration reports how many times this package has installed a queue +// adapter. It changes every time setupQueue builds a new one, which is on +// every configuration reload, and stays 0 for as long as the configuration has +// no queue section at all - in which case nothing is installed and callers are +// working with the runtime's own fallback queue. +// +// It exists because there is no way to ask for the adapter's identity from the +// outside. sdk.Runtime.GetQueueAdapter and GetQueuePrefix build a fresh +// runtime.Queue wrapper on every call, so comparing what two calls return +// compares two wrappers and never matches, however many times the underlying +// adapter has been replaced. This package creates the adapter, so this is the +// only place that knows. A counter rather than the adapter itself keeps the +// comparison on a uint64: an adapter type that is not comparable would panic +// an `==` between two interface values. +func QueueGeneration() uint64 { + queueMu.Lock() + defer queueMu.Unlock() + return installedGen +} + func setupQueue() { if config.QueueConfig.Empty() { return } - if q := sdk.Runtime.GetQueueAdapter(); q != nil { - q.Shutdown() + + queueMu.Lock() + defer queueMu.Unlock() + + // Only an adapter this package installed. GetQueueAdapter never returns + // nil - with nothing configured the runtime falls back to its own memory + // queue and wraps that - so the `if q := GetQueueAdapter(); q != nil` this + // replaces was always true, and shut down the fallback queue on the very + // first start, before anything had used it. + if installed != nil { + installed.Shutdown() } + queueAdapter, err := config.QueueConfig.Setup() if err != nil { log.Fatalf("queue setup error, %s\n", err.Error()) } sdk.Runtime.SetQueueAdapter(queueAdapter) - go queueAdapter.Run() + installed = queueAdapter + installedGen++ + + // Deliberately not started here. Run has to come after the consumers have + // registered: the contract implementations refuse a registration once the + // queue is running (storage.ErrQueueAlreadyStarted), and the legacy + // adapter this repository still goes through swallows that error rather + // than reporting it - its own comment says the interface gives it no way + // to tell the caller. Starting here and registering afterwards is + // therefore a race that loses consumers in silence. Whoever registers is + // the one that starts it. } From 94163f9afb82a26ffdf57af998e4d246f841b78a Mon Sep 17 00:00:00 2001 From: zhangwenjian Date: Sat, 5 Sep 2026 22:31:12 +0800 Subject: [PATCH 5/9] =?UTF-8?q?fix=F0=9F=90=9B:=20stop=20the=20job=20sched?= =?UTF-8?q?uler=20on=20the=20way=20out,=20and=20start=20one=20per=20tenant?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The per-tenant setup ended with `defer crontab.Stop()` on the line above `select {}`. The select never returned, so the deferred call was unreachable for the life of the process: the scheduler had never once been stopped. And because setup never returned, the `for k, db := range dbs` loop in Setup never reached its second iteration - with several tenant databases configured, only whichever one came first out of the map ever got a scheduler at all. Both fall out of deleting the select, which was blocking for nothing: cron.Start is `go c.run()` and has never needed anything to hold the caller. The stop becomes a BeforeExit callback. cron.Stop returns a context that closes once the jobs already running have finished, so the shutdown budget has something real to bound - and giving up on that wait leaves those jobs running until the process exits, which is better than holding the whole shutdown open for one job that will not end. Startup moves from a bare goroutine in run() onto AfterListen. Two reasons: the phase runs behind core's panic guard, which does not reach across a goroutine boundary, so a panic while loading jobs used to take the process down; and the jobs it starts can call the API, which is only true once the socket is accepting. It can be synchronous now precisely because setup returns. Tested where it can be: startCrontab is split out so a scheduler can be started with no database in sight. The job runs every second; after RunShutdown, two and a half seconds of silence is the assertion. The counter-proof - registering no callback, which is what this commit replaces - compiles and reports "the job fired 2 more times after shutdown". There is one test, not several, because BeforeExit closes to further registration once it has run; a second RunShutdown in the same binary would find an empty registry and pass while proving nothing. **The multi-tenant half has no test.** setup needs a *gorm.DB per tenant before it reaches the line that was blocking, and this repository's CI has no database - `make build` is CGO_ENABLED=0 with no sqlite tag. It is the same defect though: the loop could not advance past a call that never returned. --- app/jobs/jobbase.go | 32 ++++++++++++++++++++++--- app/jobs/jobbase_test.go | 52 ++++++++++++++++++++++++++++++++++++++++ cmd/api/server.go | 25 ++++++++++++++----- 3 files changed, 100 insertions(+), 9 deletions(-) create mode 100644 app/jobs/jobbase_test.go diff --git a/app/jobs/jobbase.go b/app/jobs/jobbase.go index d1f5aba9..19fab0f3 100644 --- a/app/jobs/jobbase.go +++ b/app/jobs/jobbase.go @@ -1,6 +1,7 @@ package jobs import ( + "context" "fmt" log "github.com/go-admin-team/go-admin-core/v2/logger" "github.com/go-admin-team/go-admin-core/v2/sdk" @@ -145,11 +146,36 @@ func setup(key string, db *gorm.DB) { } // 其中任务 - crontab.Start() + startCrontab(crontab) +} + +// startCrontab starts c and arranges for it to be stopped on the way out. +// +// The stop used to be `defer crontab.Stop()` followed by `select {}`. The +// select never returned, so the defer never ran and the scheduler was never +// stopped; and because setup never returned, the loop in Setup never reached +// the second tenant - only whichever database came first out of the map ever +// got a scheduler at all. cron.Start is itself `go c.run()`, so the select was +// blocking for nothing. +// +// cron.Stop returns a context that closes once the jobs already running have +// finished. That is the wait the shutdown budget exists to bound: giving up on +// it leaves those jobs running until the process exits, which is better than +// holding the whole shutdown open for one job that will not end. +func startCrontab(c *cron.Cron) { + c.Start() fmt.Println(time.Now().Format(timeFormat), " [INFO] JobCore start success.") + // 关闭任务 - defer crontab.Stop() - select {} + sdk.Runtime.SetShutdown(func(ctx context.Context) { + stopped := c.Stop() + select { + case <-stopped.Done(): + fmt.Println(time.Now().Format(timeFormat), " [INFO] JobCore stopped.") + case <-ctx.Done(): + fmt.Println(time.Now().Format(timeFormat), " [WARN] JobCore stop gave up waiting for running jobs") + } + }) } // AddJob 添加任务 AddJob(invokeTarget string, jobId int, jobName string, cronExpression string) diff --git a/app/jobs/jobbase_test.go b/app/jobs/jobbase_test.go new file mode 100644 index 00000000..0d7c0028 --- /dev/null +++ b/app/jobs/jobbase_test.go @@ -0,0 +1,52 @@ +package jobs + +import ( + "context" + "sync/atomic" + "testing" + "time" + + "github.com/go-admin-team/go-admin-core/v2/sdk" + "github.com/go-admin-team/go-admin-core/v2/sdk/pkg/cronjob" +) + +// The scheduler had never been stopped. `defer crontab.Stop()` sat directly +// above a `select {}` that never returned, so the deferred call was +// unreachable for the life of the process. +// +// There is one test rather than several because BeforeExit closes to further +// registration once it has run: a second RunShutdown in this binary would find +// an empty registry and pass while proving nothing. +func TestTheSchedulerIsStoppedOnTheWayOut(t *testing.T) { + var ticks atomic.Int64 + + c := cronjob.NewWithSeconds() + if _, err := c.AddFunc("* * * * * *", func() { ticks.Add(1) }); err != nil { + t.Fatalf("AddFunc: %v", err) + } + + startCrontab(c) + + // It has to be running before stopping it can mean anything. + deadline := time.Now().Add(5 * time.Second) + for ticks.Load() == 0 && time.Now().Before(deadline) { + time.Sleep(20 * time.Millisecond) + } + if ticks.Load() == 0 { + t.Fatal("the scheduler never ran the job, so this test cannot show it was stopped") + } + + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + if err := sdk.Runtime.RunShutdown(ctx); err != nil { + t.Fatalf("RunShutdown: %v", err) + } + + // Two and a half seconds is two more firings of a job that runs every + // second, so silence here is the assertion. + at := ticks.Load() + time.Sleep(2500 * time.Millisecond) + if n := ticks.Load() - at; n > 0 { + t.Errorf("the job fired %d more times after shutdown: the scheduler is still running", n) + } +} diff --git a/cmd/api/server.go b/cmd/api/server.go index ca510096..6829ca98 100644 --- a/cmd/api/server.go +++ b/cmd/api/server.go @@ -72,6 +72,13 @@ func setup() { // queue would have no consumers until somebody edited the config file. sdk.Runtime.SetPhase(runtime.AfterResource, attachQueueConsumers) + // On AfterListen rather than on a bare goroutine from run(). Two reasons: + // the phase runs behind core's panic guard, which does not reach across a + // goroutine boundary - a panic while loading jobs used to take the whole + // process down with a stack that named this file - and the jobs it starts + // can call the API, which is only true once the socket is accepting. + sdk.Runtime.SetPhase(runtime.AfterListen, startCronJobs) + //1. 读取配置 bootstrap.SetupConfig( file.NewSource(file.WithPath(configYml)), @@ -83,6 +90,18 @@ func setup() { log.Info(usageStr) } +// startCronJobs registers the job implementations and starts a scheduler for +// every tenant database. +// +// It is synchronous, like the phase that runs it. jobs.Setup returns now that +// the `select {}` at the end of its per-tenant setup is gone, which is what +// makes that possible; while it was there this could only be a goroutine, and +// a goroutine is outside the panic guard. +func startCronJobs() { + jobs.InitJob() + jobs.Setup(sdk.Runtime.GetAllDb()) +} + // attachedQueue is the queue generation the consumers are attached to, plus // one, so that the zero value means "attached to nothing yet". Written from // the goroutine running the phase, read from the next one - rounds never @@ -157,12 +176,6 @@ func run() error { WriteTimeout: time.Duration(config.ApplicationConfig.WriterTimeout) * time.Second, } - go func() { - jobs.InitJob() - jobs.Setup(sdk.Runtime.GetAllDb()) - - }() - if apiCheck { var routers = sdk.Runtime.GetRouter() q := sdk.Runtime.GetQueuePrefix("") From 0b78bc1e2eb6ae2f81e5849d23b9daad5bf32729 Mon Sep 17 00:00:00 2001 From: zhangwenjian Date: Sat, 5 Sep 2026 22:31:12 +0800 Subject: [PATCH 6/9] =?UTF-8?q?docs=F0=9F=93=9D:=20write=20down=20where=20?= =?UTF-8?q?the=20life-cycle=20phases=20land=20in=20this=20host?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit core's contract says what the four phases promise. This says which line of cmd/api/server.go each of them is, which is the part an application author cannot read out of core. Three things are recorded because getting them wrong is silent: - BeforeRouter is not the before registry. Those callbacks run from runStartupHooks, which is after initRouter has built the engine; the phase is before it. The two are one line apart in the same function and describing them as equivalent is the mistake this paragraph exists to prevent. - AfterResource runs again after every configuration reload, so a callback there is idempotent with respect to a resource rather than doing nothing the second time. The queue consumers are the worked example, in both directions: a new adapter must get consumers, the same adapter must not get them twice. Identity has to come from where the resource is created - GetQueueAdapter and GetQueuePrefix build a fresh wrapper per call, so two of them never compare equal however often the adapter underneath changed. - Consumers must be registered before the queue is started, and which implementation is behind the interface decides whether that matters: QueueConfig.Setup hands back queue.NewMemory when there is no redis section, and that one does not care; a redis section reaches LegacyQueueAdapter, whose Register cannot report a refusal. The table says so rather than leaving "only on redis" as a claim. What is silent is the registration alone - every publish afterwards is refused with ErrNoHandler and logged at error level, so the symptom is missing log rows plus a lot of noise, not a quiet nothing. The third-layer table loses its "queue consumers are lost after a hot reload" row, which is what AfterResource is for, and gains the honest replacement: the four phases are the only mount points there are. There is no "after the routes are installed, before the socket is listening". AfterListen is described as the port being bound rather than Serve being in its accept loop. Serve runs on another goroutine and may not have reached the first Accept; what is true is that the bind returned, so the kernel is queueing connections. A bind that fails produces no phase at all - the error returns from run() and the banner never prints. --- docs/contract.md | 87 +++++++++++++++++++++++++++++++++++++++++++++++- 1 file changed, 86 insertions(+), 1 deletion(-) diff --git a/docs/contract.md b/docs/contract.md index 0374ce60..f6233a0c 100644 --- a/docs/contract.md +++ b/docs/contract.md @@ -377,7 +377,7 @@ if res.RowsAffected == 0 { return ErrAlreadyPaid } // 别人先改了 | 应用间调用 | 零定义。A 应用要调 B 应用只能直接 import 对方的包，循环依赖就回来了 | | 领域事件 / EventBus | 无 | | 缓存的租户隔离 | `service.Service` 有 `Cache` 字段，**是否按租户隔离未验证**。当作没隔离来写 | -| 异步任务 | 有队列，但热更新后消费者会丢（issue #892） | +| 生命周期钩子之外的时点 | 只有下面那四个。没有「路由装好之后、开始监听之前」这一档 | 这几条留给后续批次，按真实需求补——现在凭空设计只会设计错。 如果你的应用卡在这里，在 issue 里说一声，那正是我们要的输入。 @@ -672,6 +672,91 @@ if sdk.Runtime.AppRoutersSealed() { /* RunAppRouters 已经跑过了 */ } --- +## 生命周期挂载点 + +除了注册路由和迁移，应用还可以把工作挂在进程生命的四个时点上，不必等宿主按名字来调自己。 +契约本身在 core，见 +[go-admin-core `docs/contract.md`](https://github.com/go-admin-team/go-admin-core/blob/main/docs/contract.md) +的「Life-cycle phases」一节。这里只写**在本仓里它们分别落在哪一行**。 + +| 阶段 | 在 `cmd/api/server.go` 的位置 | 此时可用 | +|---|---|---| +| `AfterResource` | `bootstrap.SetupConfig` 跑完 `database.Setup` / `storage.Setup` 之后 | 配置、库、缓存、队列、casbin | +| `BeforeRouter` | `initRouter()` **之前** | 以上，加引擎尚未构建这一事实 | +| `AfterListen` | `startServing()` 里，`net.Listen` 返回之后 | 全部，端口**已绑定**、连接进得来 | +| `BeforeExit` | `srv.Shutdown` 返回之后（无论它是否报错） | 全部，正在被拆掉 | + +`AfterListen` 承诺的是**端口已绑定**，不是「`Serve` 已经在 accept 循环里」—— +`srv.Serve` 在另一个 goroutine 上。这个区别是真实的：绑定成功之后内核就会把连接排进 +backlog，所以钩子里去连自己的端口不会被拒;但此刻 `Serve` 可能还没跑到第一次 `Accept`。 +绑定失败则**根本不会有这个阶段**：`net.Listen` 的错误直接从 `run()` 返回， +横幅不打印，进程非零退出。 + +```go +sdk.Runtime.SetPhase(runtime.AfterResource, func() { /* ... */ }) +sdk.Runtime.SetShutdown(func(ctx context.Context) { /* ... */ }) +``` + +### `BeforeRouter` 不等于 `before` 注册表 + +**这两个不是同一个时点，文档里别混着写。** `SetBefore` 的回调由 +`runStartupHooks()` 执行，而那是在 `initRouter()` **之后**——引擎已经建好了。 +`BeforeRouter` 在它之前。 + +顺带：`BeforeRouter` 是「硬约束：注册要赶在启动钩子之前」那一节所说的合法注册窗口之一。 +它早于 `runStartupHooks()`，所以在这里调 `sdk.Runtime.SetAppRouters` 仍然来得及。 + +### `AfterResource` 会跑很多次，回调必须扛得住 + +它在**每次配置热更新之后**都会再跑一遍，因为热更新会重建它所命名的那些资源。 +所以这里的回调要求是**「对同一个资源幂等」，不是「第二次什么都不做」**。 + +本仓自己的队列消费者就是这条规则的样板，也是它存在的理由 +（`cmd/api/server.go` 的 `attachQueueConsumers`）： + +- 热更新重建了队列适配器，挂在旧适配器上的消费者连着一个**再没人往里发消息**的队列， + 登录日志和操作日志就此停写且不出声。所以新适配器**必须**重新注册。 +- 但同一个适配器不能注册两次，否则每条消息有两个消费者，每行日志写两遍。 + +**身份不能从访问器取。** `sdk.Runtime.GetQueueAdapter()` 与 `GetQueuePrefix()` +每次调用都新造一个 `runtime.Queue` 包装，比较两次返回等于比较两个包装， +**底层适配器换过多少次都不相等**。要在**创建资源的地方**记身份—— +本仓是 `common/storage.QueueGeneration()`。 + +### 注册消费者要赶在队列启动之前 + +走哪条实现，取决于配置里有没有 `redis:` 段（`config.QueueConfig.Setup()`）： + +| 配置 | 实际类型 | 启动后还能注册吗 | +|---|---|---| +| `queue: memory:` | `queue.NewMemory` | **能**。它的 `Register` 每次起一个消费 goroutine，不看是否已 `Run` | +| `queue: redis:` | `storage.LegacyQueueAdapter` 包着新契约实现 | **不能**。`Register` 内部调 `Subscribe`，启动后返回 `storage.ErrQueueAlreadyStarted` | + +而 `LegacyQueueAdapter.Register` **没有返回值**——它只能把这个错误写进 slog， +core 里那行注释自己写着「The interface has no way to report this to the caller」。 +**静默的是注册这一步，不是之后。** 没有建立消费组，redis 会用 +`storage.ErrNoHandler` 拒绝**之后的每一次投递**，而本仓两个调用点 +（`common/middleware/logger.go`、`common/middleware/handler/auth.go`） +都把它记为 error——于是日志行一条都不落库，同时每个请求刷一条错误日志。 + +所以顺序是硬的：**先 `Register` 完，再由注册方 `Run()`。** +`common/storage` 的 `setupQueue` 有意不启动队列。 + +默认配置选的是 memory 后端,它不在乎顺序——**这个缺陷在默认部署里看不见, +只在配了 redis 的部署上发作**,而丢掉的正是登录日志、操作日志和 api 检查。 + +### `BeforeExit` 反序执行，预算约束的是等待 + +清理按**注册的逆序**执行。`SetShutdown` 拿到宿主剩余的预算， +但**它约束的是等待，不是工作**：预算用尽时 `RunShutdown` 停止等待并返回， +而不检查 context 的回调会一直跑到进程退出。Go 没法取消一个不检查取消的函数。 + +本仓的样板是 cron（`app/jobs/jobbase.go` 的 `startCrontab`）： +`cron.Stop()` 返回一个在**已经在跑的任务结束时**关闭的 context， +钩子在它和预算之间二选一。 + +--- + ## 安全边界：装一个应用等于信任它 **这一层划不出安全边界，本文不假装划得出。** From dff0e64f510819bf507fa6036ab87e3518c54414 Mon Sep 17 00:00:00 2001 From: zhangwenjian Date: Sat, 5 Sep 2026 22:31:12 +0800 Subject: [PATCH 7/9] =?UTF-8?q?test=E2=9C=85:=20run=20the=20queue's=20orde?= =?UTF-8?q?ring=20rule=20against=20a=20real=20redis=20in=20CI?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The rule that consumers are registered before the queue is started had no test that could fail on the backend it exists for. Everything so far ran on the memory queue, which is the default: queue.Memory's Register starts another consumer goroutine whatever the state, so the wrong order passes there. A suite that only exercises the default reports success for a queue that accepts no consumers at all. CI gets a redis service, and two tests build the queue the way setupQueue does - through config.QueueConfig.Setup, so what is under test is the adapter this repository actually gets, LegacyQueueAdapter included. They skip without GO_ADMIN_TEST_REDIS_ADDR, so a developer with no server still gets a green run. Registering first and starting second delivers the message. Starting first and registering second is refused: no consumer group was created, so Append comes back with storage.ErrNoHandler. Pinning that particular error rather than "some error" is deliberate - the test is about the missing consumer, and a connection failure that happened to error too would otherwise pass for it. Running it corrected something written two commits ago. The claim there was that a late registration loses consumers "with nothing said". Only half of that holds: the registration is silent, because Register returns nothing, but every publish afterwards fails loudly - ErrNoHandler, logged at error level by both call sites in common/middleware - while the log rows are never written. The symptom is missing rows plus a lot of noise, not a quiet nothing. That commit's message and the contract doc both say so now. --- .github/workflows/go.yml | 21 +++++ common/storage/queue_redis_test.go | 122 +++++++++++++++++++++++++++++ 2 files changed, 143 insertions(+) create mode 100644 common/storage/queue_redis_test.go diff --git a/.github/workflows/go.yml b/.github/workflows/go.yml index 601080c6..8872dc13 100644 --- a/.github/workflows/go.yml +++ b/.github/workflows/go.yml @@ -15,6 +15,27 @@ jobs: build: name: Build runs-on: ubuntu-latest + + # The queue's ordering rule - consumers registered before the queue is + # started - is invisible on the memory backend, which is the default and + # therefore what every other test runs on: queue.Memory's Register starts a + # consumer goroutine whatever the state. Only redis refuses a late + # registration, so without a server here the tests that cover it would skip + # and the suite would report success for a queue that accepts no consumers. + services: + redis: + image: redis:7-alpine + ports: + - 6379:6379 + options: >- + --health-cmd "redis-cli ping" + --health-interval 5s + --health-timeout 3s + --health-retries 10 + + env: + GO_ADMIN_TEST_REDIS_ADDR: 127.0.0.1:6379 + steps: - name: Set up Go 1.26 diff --git a/common/storage/queue_redis_test.go b/common/storage/queue_redis_test.go new file mode 100644 index 00000000..ca14b120 --- /dev/null +++ b/common/storage/queue_redis_test.go @@ -0,0 +1,122 @@ +package storage + +import ( + "errors" + "os" + "testing" + "time" + + "github.com/go-admin-team/go-admin-core/v2/sdk/config" + corestorage "github.com/go-admin-team/go-admin-core/v2/storage" + "github.com/go-admin-team/go-admin-core/v2/storage/queue" +) + +// redisAddrEnv points these tests at a server. They are skipped without it, so +// a developer with no redis running still gets a green run - and CI sets it, +// which is the point: the ordering rule they cover is invisible on the memory +// backend, and memory is the default. A suite that only ever exercised the +// default would report success for a queue that silently drops every consumer. +const redisAddrEnv = "GO_ADMIN_TEST_REDIS_ADDR" + +func redisAddr(t *testing.T) string { + t.Helper() + addr := os.Getenv(redisAddrEnv) + if addr == "" { + t.Skipf("%s is not set; skipping the redis-backed queue tests", redisAddrEnv) + } + return addr +} + +// newRedisQueue builds the queue the same way setupQueue does - through +// config.QueueConfig.Setup - so that what is under test is the adapter this +// repository actually gets, LegacyQueueAdapter and all, rather than a redis +// client wired up by the test. +func newRedisQueue(t *testing.T, prefix string) corestorage.AdapterQueue { + t.Helper() + previous := config.QueueConfig + t.Cleanup(func() { config.QueueConfig = previous }) + + config.QueueConfig = &config.Queue{ + Redis: &config.RedisQueue{ + RedisOptions: config.RedisOptions{Addr: redisAddr(t)}, + Group: prefix, + KeyPrefix: prefix, + }, + } + q, err := config.QueueConfig.Setup() + if err != nil { + t.Fatalf("queue setup: %v", err) + } + t.Cleanup(q.Shutdown) + return q +} + +func message(t *testing.T, stream string) corestorage.Messager { + t.Helper() + m := &queue.Message{} + m.SetStream(stream) + m.SetValues(map[string]interface{}{"hello": "world"}) + return m +} + +// Registered first, then started: the consumer gets the message. This is the +// order setupQueue and attachQueueConsumers now produce between them. +func TestRedisQueueDeliversToAConsumerRegisteredBeforeTheStart(t *testing.T) { + stream := "t-ordered" + q := newRedisQueue(t, "gotest-ordered") + + got := make(chan struct{}, 1) + q.Register(stream, func(corestorage.Messager) error { + select { + case got <- struct{}{}: + default: + } + return nil + }) + go q.Run() + + // Give Start a moment to reach its read loop before publishing. + time.Sleep(500 * time.Millisecond) + if err := q.Append(message(t, stream)); err != nil { + t.Fatalf("append: %v", err) + } + + select { + case <-got: + case <-time.After(15 * time.Second): + t.Fatal("the consumer never received the message") + } +} + +// Started first, then registered: the registration is refused and every +// publish afterwards fails. +// +// Subscribe answers ErrQueueAlreadyStarted, and LegacyQueueAdapter.Register +// returns nothing, so the caller cannot know - that part is silent. What is not +// silent is the consequence: no consumer group was created, so Publish refuses +// the topic with ErrNoHandler on every single request, and go-admin's call +// sites log that at error level while the login and operation log rows are +// never written. +// +// This is the test the memory backend cannot provide. queue.Memory's Register +// starts another consumer goroutine whatever the state, so the same code passes +// there - which is how the defect survived, memory being the default. +func TestRedisQueueRefusesAConsumerRegisteredAfterTheStart(t *testing.T) { + stream := "t-late" + q := newRedisQueue(t, "gotest-late") + + go q.Run() + time.Sleep(500 * time.Millisecond) + + q.Register(stream, func(corestorage.Messager) error { return nil }) + + err := q.Append(message(t, stream)) + if err == nil { + t.Fatal("a message was accepted for a topic whose registration came after Start; " + + "if the backend now accepts late registration, the ordering rule in setupQueue can be revisited") + } + if !errors.Is(err, corestorage.ErrNoHandler) { + t.Fatalf("append failed with %v, want %v - the test is meant to pin the "+ + "missing-consumer path, not any error at all", err, corestorage.ErrNoHandler) + } +} From 36f2549172897670fda27d08208547afb7ce692d Mon Sep 17 00:00:00 2001 From: zhangwenjian Date: Sat, 5 Sep 2026 22:31:12 +0800 Subject: [PATCH 8/9] =?UTF-8?q?test=E2=9C=85:=20pin=20BeforeRouter=20to=20?= =?UTF-8?q?the=20moment=20before=20the=20engine=20exists?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit buildRouter is split out of run() so the order it establishes can be asserted: the phase is announced, then initRouter builds the engine, then runStartupHooks drains the registries. The test covers both halves of the distinction the contract draws. A BeforeRouter callback sees no engine - that is what the phase means, the last point at which a module can still affect how routes are built. A callback in the before registry, two lines later, sees one. The names invite treating them as the same moment and they are not. No database is involved. AuthInit reads ApplicationConfig.Mode and JwtConfig and nothing else, and building a router registers handlers rather than calling them, so the whole sequence runs in a package test with two package-level values set. freshRuntime swaps the global runtime for the duration: runStartupHooks seals the registries it drains, and a sealed registry silently drops everything registered afterwards, which would leave every later test in this binary passing while proving nothing. The counter-proof compiles and fails - announcing the phase after initRouter reports "BeforeRouter saw engine &{...}, want nil". --- cmd/api/phase_test.go | 58 +++++++++++++++++++++++++++++++++++++++++++ cmd/api/server.go | 25 ++++++++++++------- 2 files changed, 74 insertions(+), 9 deletions(-) diff --git a/cmd/api/phase_test.go b/cmd/api/phase_test.go index 20d5b80c..aa28672e 100644 --- a/cmd/api/phase_test.go +++ b/cmd/api/phase_test.go @@ -8,6 +8,7 @@ import ( "time" "github.com/go-admin-team/go-admin-core/v2/sdk" + "github.com/go-admin-team/go-admin-core/v2/sdk/config" "github.com/go-admin-team/go-admin-core/v2/sdk/runtime" ) @@ -86,3 +87,60 @@ func TestAfterListenIsAnnouncedOnlyOnceThePortIsBound(t *testing.T) { } _ = c.Close() } + +// BeforeRouter is the last point at which a module can still affect how routes +// are built, so it has to run while there is no engine yet. The before registry +// is a different moment despite the name: those callbacks run after initRouter +// has built the engine. +// +// The two are two lines apart in buildRouter, and calling them equivalent is a +// mistake this repository has already made in writing. Until this test the +// ordering was checked by reading - which is how the stop signals came to be +// armed after the readiness banner in the same file. +func TestBeforeRouterRunsWhileThereIsNoEngine(t *testing.T) { + freshRuntime(t) + + // AuthInit reads these two package-level values and nothing else. No + // database is involved in building a router: the handlers are registered, + // not called. + config.ApplicationConfig.Mode = "dev" + config.JwtConfig.Secret = "test-secret-for-the-router-build" + + type observation struct { + ran int + engineWas interface{} + engineSeen bool + } + var phase, before observation + + sdk.Runtime.SetPhase(runtime.BeforeRouter, func() { + phase.ran++ + phase.engineWas = sdk.Runtime.GetEngine() + phase.engineSeen = true + }) + sdk.Runtime.SetBefore(func() { + before.ran++ + before.engineWas = sdk.Runtime.GetEngine() + before.engineSeen = true + }) + + buildRouter() + + if phase.ran != 1 { + t.Fatalf("BeforeRouter ran %d times, want 1", phase.ran) + } + if !phase.engineSeen || phase.engineWas != nil { + t.Errorf("BeforeRouter saw engine %v, want nil: it is meant to run before initRouter builds one", phase.engineWas) + } + + if before.ran != 1 { + t.Fatalf("the before registry ran %d times, want 1", before.ran) + } + if before.engineWas == nil { + t.Error("a before callback saw no engine; that registry is meant to run after initRouter, and describing it as equivalent to BeforeRouter is the error this asserts against") + } + + if sdk.Runtime.GetEngine() == nil { + t.Error("buildRouter returned with no engine built") + } +} diff --git a/cmd/api/server.go b/cmd/api/server.go index 6829ca98..9c242f56 100644 --- a/cmd/api/server.go +++ b/cmd/api/server.go @@ -159,15 +159,7 @@ func run() error { if config.ApplicationConfig.Mode == pkg.ModeProd.String() { gin.SetMode(gin.ReleaseMode) } - // The last point at which a module can still affect how routes are built. - // It is not the same moment as the before registry, which runStartupHooks - // drains below - those callbacks run after initRouter has built the engine, - // not before it. - sdk.Runtime.RunPhase(runtime.BeforeRouter) - - initRouter() - - runStartupHooks() + buildRouter() srv := &http.Server{ Addr: fmt.Sprintf("%s:%d", config.ApplicationConfig.Host, config.ApplicationConfig.Port), @@ -342,6 +334,21 @@ func runShutdownHooks(timeout time.Duration) error { return sdk.Runtime.RunShutdown(ctx) } +// buildRouter announces BeforeRouter, builds the engine, and then drains the +// startup registries. +// +// The order is the contract. BeforeRouter is the last point at which a module +// can still affect how routes are built, so it has to run while there is no +// engine yet. The before registry runStartupHooks drains is a different moment +// despite the name: those callbacks run after initRouter has built the engine. +// Two lines apart, and describing them as equivalent is a mistake this +// repository has already made once in writing. +func buildRouter() { + sdk.Runtime.RunPhase(runtime.BeforeRouter) + initRouter() + runStartupHooks() +} + // runStartupHooks runs the router registries and then the before callbacks. // // The package-level slice runs first and in its existing order, so a fork that From 8ee4141af6af3af5729f863b8874e2d68efab39d Mon Sep 17 00:00:00 2001 From: zhangwenjian Date: Sat, 5 Sep 2026 22:41:53 +0800 Subject: [PATCH 9/9] =?UTF-8?q?fix=F0=9F=90=9B:=20check=20the=20certificat?= =?UTF-8?q?e=20before=20announcing=20that=20the=20port=20is=20reachable?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit AfterListen promises a hook that the port answers. The bind was moved onto the caller's goroutine to keep that promise, but with ssl enabled there was a second way to fail after the announcement: ServeTLS reads the certificate files itself, on the serving goroutine, so a bad path or an unreadable key surfaced once the hooks had already run. tls.LoadX509KeyPair now runs before anything is announced, and its error is returned from startServing the way a failed bind is. ServeTLS still does the real work - handing it a tls.Listener built here instead would take over the HTTP/2 negotiation it sets up, and quietly drop h2 for every TLS deployment. The cost is one extra read of the certificate at startup. The fatal in the serving goroutine said "listen:". Neither the bind nor the certificate reaches it any more, so it says "serve:". The test covers the certificate path alongside the bind: neither may announce the phase, and neither may seal it. The counter-proof is not clean, and saying so is the point. Removing the check does turn the run red, but through log.Fatal killing the process from the serving goroutine - "fatal serve: open no-such.pem: no such file or directory" - rather than through the assertion. That still demonstrates the defect, because the process could only get there after startServing had returned successfully and the phase had been announced; it cannot be observed from inside the test, because the fatal races the assertion that would report it. Also: the redis-backed queue tests now fail instead of skipping when CI is set and GO_ADMIN_TEST_REDIS_ADDR is not. A workflow that renamed the variable or dropped the service would otherwise stay green while those two tests quietly did nothing - the same shape as the defect they exist to cover. Locally, with no CI in the environment, they still skip. --- cmd/api/phase_test.go | 14 +++++++++++++ cmd/api/server.go | 33 ++++++++++++++++++++++-------- common/storage/queue_redis_test.go | 14 ++++++++++--- 3 files changed, 50 insertions(+), 11 deletions(-) diff --git a/cmd/api/phase_test.go b/cmd/api/phase_test.go index aa28672e..c63d491c 100644 --- a/cmd/api/phase_test.go +++ b/cmd/api/phase_test.go @@ -63,6 +63,20 @@ func TestAfterListenIsAnnouncedOnlyOnceThePortIsBound(t *testing.T) { t.Error("a failed bind sealed AfterListen, so the phase could never run for a server that did start") } + // A certificate that cannot be read is the other way to fail before there + // is anything to announce. ServeTLS reads it on the serving goroutine, so + // without the check in startServing this would be a hook told the port was + // reachable while the server was already on its way down. + if err := startServing(&http.Server{Addr: "127.0.0.1:0"}, true, "no-such.pem", "no-such.key"); err == nil { + t.Fatal("startServing returned no error for a certificate that does not exist") + } + if ran != 0 { + t.Errorf("AfterListen ran %d times after a certificate failure", ran) + } + if sdk.Runtime.PhaseSealed(runtime.AfterListen) { + t.Error("a certificate failure sealed AfterListen") + } + // And now a bind that works. port := freePort(t) srv := &http.Server{Addr: fmt.Sprintf("127.0.0.1:%d", port), Handler: http.NewServeMux()} diff --git a/cmd/api/server.go b/cmd/api/server.go index 9c242f56..2237ac76 100644 --- a/cmd/api/server.go +++ b/cmd/api/server.go @@ -2,6 +2,7 @@ package api import ( "context" + "crypto/tls" "fmt" "net" "net/http" @@ -280,7 +281,25 @@ func armStopSignals() (<-chan os.Signal, func()) { // AfterListen is announced synchronously. Running it in a goroutine to save the // few milliseconds would let it overlap the shutdown: on a fast SIGTERM the // cleanup callbacks could finish before the startup ones had. +// +// Both ways of failing to start are therefore checked before the announcement: +// the bind, and - with ssl enabled - the certificate. func startServing(srv *http.Server, useTLS bool, pem, key string) error { + if useTLS { + // Read the certificate before anything is announced. ServeTLS reads + // these files itself, but on the serving goroutine - so a bad + // certificate used to surface after AfterListen had already promised a + // reachable port. Loading it here costs one extra read and moves the + // failure onto this goroutine, where run() can return it. + // + // ServeTLS still does the real work below rather than this handing it a + // tls.Listener: that is what sets up HTTP/2 negotiation, and taking it + // over here would quietly drop h2 for every TLS deployment. + if _, err := tls.LoadX509KeyPair(pem, key); err != nil { + return errors.Wrap(err, "tls certificate") + } + } + ln, err := net.Listen("tcp", srv.Addr) if err != nil { return errors.Wrap(err, "listen") @@ -295,14 +314,12 @@ func startServing(srv *http.Server, useTLS bool, pem, key string) error { err = srv.Serve(ln) } if err != nil && !errors.Is(err, http.ErrServerClosed) { - // Still fatal, as it was. The bind is no longer among the errors - // that reach here; what is left is a serve that failed after the - // port was taken, and carrying on would park the process on - // <-quit with nothing serving. TLS is the one case that can still - // fail immediately - ServeTLS reads the certificate files - so - // with ssl enabled AfterListen can run against a server that is - // already on its way down. - log.Fatal("listen: ", err) + // Still fatal, as it was. Neither the bind nor the certificate is + // among the errors that reach here any more - both are checked + // above, on the caller's goroutine. What is left is a serve that + // failed after the port was taken, and carrying on would park the + // process on <-quit with nothing serving. + log.Fatal("serve: ", err) } }() diff --git a/common/storage/queue_redis_test.go b/common/storage/queue_redis_test.go index ca14b120..4a7311a4 100644 --- a/common/storage/queue_redis_test.go +++ b/common/storage/queue_redis_test.go @@ -21,10 +21,18 @@ const redisAddrEnv = "GO_ADMIN_TEST_REDIS_ADDR" func redisAddr(t *testing.T) string { t.Helper() addr := os.Getenv(redisAddrEnv) - if addr == "" { - t.Skipf("%s is not set; skipping the redis-backed queue tests", redisAddrEnv) + if addr != "" { + return addr } - return addr + // Skipping locally is the point; skipping in CI is the failure this whole + // file exists to prevent. A workflow that renamed the variable, or dropped + // the service, would otherwise go green while these two tests quietly did + // nothing - which is the same shape as the defect they cover. + if os.Getenv("CI") != "" { + t.Fatalf("%s is not set while CI is: the redis-backed queue tests must not skip here", redisAddrEnv) + } + t.Skipf("%s is not set; skipping the redis-backed queue tests", redisAddrEnv) + return "" } // newRedisQueue builds the queue the same way setupQueue does - through