Compare commits

..
Author SHA1 Message Date
zhangwenjian 211ae85a4e test✅: fail on an Append error this test does not model
Only ErrQueueClosed was examined and every other error was discarded, so a
run where nothing reached a queue at all would leave the refusal count at
zero and the test green. The first unexpected error is now kept and fails
the test.

The pool is sized so a full queue cannot be one of them: it returns an error
of its own and is expected while the consumer is held, which would otherwise
make the new check fire on the normal path.
2026-09-06 19:00:30 +08:00
zhangwenjian 3c3d94ca76 test✅: wait for the sample rather than for the clock
Two fixed sleeps decided when this test looked: one to let the consumer pick
a message up, one to let publishes accumulate inside the reload. Both are
guesses about how fast the runner is. The consumer now signals on its first
delivery, and the measurement waits until enough publishes have landed.

The "nothing was published" guard goes with them. It was the weaker form of
the same check, and it ran after the fact instead of holding the window open
until there was something to measure.
2026-09-06 19:00:25 +08:00
zhangwenjian 6138d2d74c test✅: assert the bound installing first actually gives
The assertion demanded zero refusals during a reload, and this ordering
cannot deliver that. GetQueuePrefix returns a wrapper that captured the
adapter, so a producer that fetched before the swap and appends after
Shutdown has begun is still holding the old queue. That window is one call
wide; closing it means resolving the adapter inside Append, which is core's
to change.

What the ordering removes is the sustained window - every producer that
fetches during the wait. Measured on a race-enabled run: 174 of 174 publishes
refused with the old order, 1 of 174 with the new one. The old assertion
therefore failed about half the time on a change that works.
2026-09-06 19:00:20 +08:00
zhangwenjian d5de79f75b test✅: join the producer before the test returns
The publishing goroutine was told to stop and never waited for. t.Cleanup
restores sdk.Runtime while a producer that has not yet noticed the stop is
still reading it, which -race reports as a write and a read on the same
package variable. Signalling is not joining.
2026-09-06 19:00:10 +08:00
zhangwenjian 46e793972c fix🐛: install the new queue before shutting the old one down
Shutdown now waits for its consumers to deliver what the queue still holds, and
setupQueue called it first. For that whole wait sdk.Runtime still pointed at the
adapter that had stopped accepting, so every Append landing in the window came
back ErrQueueClosed - and both call sites in common/middleware log that at error
level while the row never reaches the database.

Measured with a held consumer: 177 of 178 publishes during one reload.

Installing first leaves no window. A producer fetches the adapter per call and
gets either the new queue or the old one, and both accept; the old one still
drains, because Shutdown is what waits for that.

The difference only exists during the wait - after Setup returns the two orders
look identical, which is why the test holds a consumer and publishes throughout
the reload rather than checking the state afterwards.

The counter-proof compiles and reports the 177.
2026-09-06 17:37:01 +08:00
zhangwenjian 46f4092b43 build🔧: require go-admin-core v2.7.0
It carries the queue shutdown fixes: a closed queue now delivers what it already
accepted and stops the goroutines consuming it. Both matter here, because this
host rebuilds its queue adapter on every configuration reload and shuts one down
on the way out.

Nothing in this commit uses the new behaviour. The one place that has to change
because of it follows.
2026-09-06 17:37:01 +08:00
wenjianzhang 196195357b Merge pull request #908 from go-admin-team/feat/readiness-probe
feat✨: 新增 /ready,并让它在关闭一开始就失败
2026-09-06 10:07:45 +08:00
zhangwenjian c579c5f84c fix🐛: give every cache probe its own key
One fixed key meant two probes overlapping - two /ready requests, or two
instances against the one redis, which is the normal deployment - overwrote each
other's value between the write and the read, and each concluded the cache was
broken. A readiness probe that reports false negatives under load pulls healthy
instances out of the pool, which is worse than not probing.

The key now carries eight random bytes. The written value is still read back,
because that is what tells a healthy cache apart from one that answers "miss"
for everything, and the key is deleted afterwards on a best-effort basis - its
error is dropped deliberately, since the verdict is already decided and a cache
that cannot delete what it just wrote is not a reason to refuse traffic.

The first version of the concurrency test had no teeth: the probes are short
enough that the scheduler ran them one after another, so the fixed-key
counter-proof passed three times out of three. The fake cache now holds every
writer until all of them have written, which makes the interleaving the test is
about actually happen. With one key that is a deterministic 15 failures out of
16 - only the last writer's value survives - and with a key per probe, none.
2026-09-06 09:55:12 +08:00
zhangwenjian 241c27358b feat✨: answer readiness separately from liveness, and fail it while draining
/health returned 200 without asking anything. Whatever it was meant to say, an
orchestrator reading it learned only that a process was accepting connections.

The two questions are not the same one, and the answers differ:

  - /health stays a bare 200. It answers "should I restart you", and a process
    whose database is unreachable does not want restarting - that turns one
    outage into a crash loop and discards the connection pool, the cache and
    every request in flight on the way.
  - /ready is new. It answers "should I send you requests", fails while a
    dependency is unreachable, and fails from the moment shutdown begins.

That last part is what the life-cycle phases bought. BeginDraining sits next to
BeginShutdown, before the server stops accepting, so a load balancer is told to
stop sending while this instance can still finish what it holds. Reversed - and
that is where it was - the connections are cut first and the probe reports it
afterwards.

The queue is deliberately not checked. Nothing on AdapterQueue answers "are you
reachable" without publishing something, the memory backend cannot fail, and a
queue that is down degrades logging rather than stopping requests: a reason to
alert, not a reason to leave the pool.

The cache probe writes and reads back rather than only reading. A cache that
answers "miss" for every key - a client pointed at the wrong server - is
indistinguishable from a healthy one on a read alone.

Every check runs behind a recover, and that is not defensive habit. The test for
"nothing configured" found the reason: GetCacheAdapter builds a wrapper around
whatever is configured and returns it even when nothing is, so the value is not
nil, the cache inside it is, and Set dereferences it. A nil check cannot see
that, and GetQueueAdapter behaves the same way. Whatever the cause, a probe is
the last thing that should be able to take the process down - the caller is
asking whether this instance is well, and killing it is the wrong reply.

The counter-proof compiles and fails: without the recover, the unconfigured case
panics rather than reporting two failed checks.
2026-09-06 09:47:55 +08:00
wenjianzhang aa3c9866cb Merge pull request #907 from go-admin-team/test/892-reload-generation
test✅: 补上 #892 修复中没有断言的那一半
2026-09-06 09:47:07 +08:00
zhangwenjian 2ac01ea584 test✅: restore what Setup writes outside this test
Setup installs the cache and queue adapters on sdk.Runtime, which is a
package-level singleton, and records what it installed in this package's own
variables. The test left all of it behind, so what a later test in this binary
saw depended on whether this one had run - the leak cmd/api's freshRuntime and
common/middleware's copy of it exist to prevent.

sdk.Runtime is swapped for a fresh one and everything is put back in Cleanup.
2026-09-05 23:18:01 +08:00
zhangwenjian 7f9cc1e435 test✅: a configuration reload installs a new queue for the consumers to notice
Issue #892 is that a reload replaces the queue adapter and the consumers
registered against the previous one are left attached to a queue nobody
publishes to any more.

The fix has two halves and only one of them was covered. attachQueueConsumers
gives a new queue its own consumers and the same queue none, which cmd/api
tests against a queue it controls by passing generation numbers in by hand. What
nothing asserted is that a reload actually produces a new generation for it to
notice - the half that lives in this package.

Setup is what config re-runs on every change, so calling it twice is what a
reload does here. The generation goes 0, 1, 2.

The counter-proof compiles and fails: dropping the increment in setupQueue
reports that the first Setup installed no queue at all, because a generation
that never moves is indistinguishable from never having been set.

This closes the loop rather than adding coverage for its own sake: with both
halves asserted, the claim that #892 is fixed rests on tests instead of on
reading the two functions and believing they meet.
2026-09-05 23:01:27 +08:00
wenjianzhang 4fb0529d2d Merge pull request #906 from go-admin-team/feat/005-host-wiring
feat✨: 生命周期阶段接线,并修掉队列注册顺序与从未停止的 cron
2026-09-05 22:49:50 +08:00
zhangwenjian 8ee4141af6 fix🐛: check the certificate before announcing that the port is reachable
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.
2026-09-05 22:41:53 +08:00
zhangwenjian 36f2549172 test✅: pin BeforeRouter to the moment before the engine exists
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".
2026-09-05 22:31:12 +08:00
zhangwenjian dff0e64f51 test✅: run the queue's ordering rule against a real redis in CI
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.
2026-09-05 22:31:12 +08:00
zhangwenjian 0b78bc1e2e docs📝: write down where the life-cycle phases land in this host
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.
2026-09-05 22:31:12 +08:00
zhangwenjian 94163f9afb fix🐛: stop the job scheduler on the way out, and start one per tenant
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.
2026-09-05 22:31:12 +08:00
zhangwenjian 4510b06959 fix🐛: register the queue consumers before the queue is started
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.
2026-09-05 22:31:12 +08:00
zhangwenjian 750c7c744e feat✨: run the BeforeExit callbacks on the way out
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.
2026-09-05 22:31:12 +08:00
zhangwenjian d52dca1cb6 feat✨: announce BeforeRouter and AfterListen, and bind before either
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.
2026-09-05 22:31:11 +08:00
zhangwenjian 71413a4248 build🔧: require go-admin-core v2.6.0
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".
2026-09-05 22:31:11 +08:00
wenjianzhang 4fede43254 Merge pull request #905 from go-admin-team/fix/checksilent-datascope-route
feat✨(checksilent): 抓「handler 读数据权限、路由却没挂中间件」
2026-09-05 21:56:08 +08:00
wenjianzhang 37065fb089 Merge pull request #904 from go-admin-team/fix/getinfo-data-permission
fix🐛: 开了数据权限就登不进去——/getinfo 拿不到它要的 DataPermission
2026-09-05 21:34:13 +08:00
18 changed files with 1663 additions and 52 deletions
+21
View File
@@ -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
+29 -3
View File
@@ -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)
+52
View File
@@ -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)
}
}
+43 -3
View File
@@ -1,23 +1,63 @@
package router
import (
"context"
"net/http"
"time"
"github.com/gin-gonic/gin"
"github.com/go-admin-team/go-admin-core/v2/tools/transfer"
"github.com/prometheus/client_golang/prometheus/promhttp"
"go-admin/common/health"
)
func init() {
routerNoCheckRole = append(routerNoCheckRole, registerMonitorRouter)
}
// 需认证的路由代码
// readyTimeout bounds the whole probe. It has to stay under whatever period
// the orchestrator polls on, or a slow dependency turns a readiness check into
// a queue of readiness checks.
const readyTimeout = 2 * time.Second
// 无需认证的路由代码
func registerMonitorRouter(v1 *gin.RouterGroup) {
v1.GET("/metrics", transfer.Handler(promhttp.Handler()))
//健康检查
// 健康检查(存活)
//
// Stays a bare 200 on purpose. This is the answer to "should I restart
// you", and a process whose database is unreachable does not want
// restarting - that turns one outage into a crash loop and throws away the
// connection pool, the cache and every in-flight request along the way.
v1.GET("/health", func(c *gin.Context) {
c.Status(http.StatusOK)
})
}
// 就绪检查
//
// The answer to "should I send you requests". It fails while a dependency
// is unreachable, and from the moment shutdown begins - which is before
// the server stops accepting, so a load balancer can take this instance
// out of the pool while it can still finish what it has.
v1.GET("/ready", func(c *gin.Context) {
if health.Draining() {
c.JSON(http.StatusServiceUnavailable, gin.H{
"status": "draining",
"checks": []health.Check{},
})
return
}
ctx, cancel := context.WithTimeout(c.Request.Context(), readyTimeout)
defer cancel()
checks := health.Ready(ctx)
status := http.StatusOK
if !health.Healthy(checks) {
status = http.StatusServiceUnavailable
}
c.JSON(status, gin.H{"status": http.StatusText(status), "checks": checks})
})
}
+160
View File
@@ -0,0 +1,160 @@
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/config"
"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")
}
// 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()}
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()
}
// 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")
}
}
+167
View File
@@ -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())
}
}
+209 -32
View File
@@ -2,10 +2,13 @@ package api
import (
"context"
"crypto/tls"
"fmt"
"net"
"net/http"
"os"
"os/signal"
"sync/atomic"
"syscall"
"time"
@@ -14,8 +17,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"
@@ -24,6 +30,7 @@ import (
"go-admin/app/jobs"
"go-admin/common/database"
"go-admin/common/global"
"go-admin/common/health"
common "go-admin/common/middleware"
"go-admin/common/middleware/handler"
"go-admin/common/storage"
@@ -60,30 +67,101 @@ 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)
// 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. 读取配置
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)
}
// 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
// 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)
}
initRouter()
runStartupHooks()
buildRouter()
srv := &http.Server{
Addr: fmt.Sprintf("%s:%d", config.ApplicationConfig.Host, config.ApplicationConfig.Port),
@@ -92,12 +170,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("")
@@ -122,18 +194,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:"))
@@ -149,6 +213,18 @@ 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()
// Readiness fails from here, which is before the server stops accepting.
// The order is the whole point: a load balancer that is told "not ready"
// while this instance can still finish what it has in flight takes it out
// of the pool without dropping anything. Reversed, the connections are cut
// first and the health check reports it afterwards.
health.BeginDraining()
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 +232,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.
@@ -185,6 +274,66 @@ 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.
//
// 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")
}
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. 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)
}
}()
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
@@ -196,6 +345,34 @@ 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)
}
// 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
+84 -7
View File
@@ -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)
}
}
+157
View File
@@ -0,0 +1,157 @@
// Package health answers whether this process should be sent traffic.
//
// The two questions an orchestrator asks are not the same one, and go-admin
// answers them at two endpoints:
//
// - /health is liveness: is the process there at all. It stays a bare 200,
// because the honest answer to "should I restart you" is almost always no.
// Restarting a process because its database is unreachable turns one
// outage into a crash loop that also loses the connection pool, the cache
// and every in-flight request.
// - /ready is readiness: should this instance receive requests now. It fails
// while the dependencies are unreachable, and - the part that only exists
// because of the life-cycle phases - it fails as soon as shutdown begins,
// before the server stops accepting, so a load balancer has a chance to
// take the instance out before connections are cut.
package health
import (
"context"
"crypto/rand"
"encoding/hex"
"errors"
"fmt"
"sync/atomic"
"time"
"github.com/go-admin-team/go-admin-core/v2/sdk"
)
// draining is set when the process starts shutting down.
//
// It is kept here rather than read back from core: BeginShutdown sets a flag on
// the Application, but nothing exports it, and one host wanting to know is not
// yet a reason to widen that interface.
var draining atomic.Bool
// BeginDraining records that shutdown has started, so readiness fails from now
// on. It is called with BeginShutdown, before anything is taken apart.
func BeginDraining() { draining.Store(true) }
// Draining reports whether shutdown has begun.
func Draining() bool { return draining.Load() }
// Check is one dependency and what asking it produced.
type Check struct {
Name string `json:"name"`
OK bool `json:"ok"`
Err string `json:"error,omitempty"`
}
// Ready asks every dependency this process cannot serve a request without.
//
// The queue is deliberately absent. Nothing on AdapterQueue answers "are you
// reachable" without publishing something, the memory backend cannot fail, and
// a queue that is down degrades logging rather than stopping requests - which
// is a reason to alert, not a reason to leave the load balancer pool.
func Ready(ctx context.Context) []Check {
return []Check{
safely("database", func() error { return pingDB(ctx) }),
safely("cache", probeCache),
}
}
// safely turns a panic into a failed check.
//
// Not defensive habit: the accessors hand back wrappers, not the resources.
// sdk.Runtime.GetCacheAdapter builds a runtime.Cache around whatever is
// configured and returns it even when nothing is - so the value is not nil, the
// cache inside it is, and the first call dereferences it. A nil check cannot
// see that, and the same is true of GetQueueAdapter.
//
// Whatever the reason, a probe is the last thing that should be able to take
// the process down: the caller is asking whether this instance is well, and
// killing it to answer is the wrong reply.
func safely(name string, fn func() error) (c Check) {
c = Check{Name: name}
defer func() {
if r := recover(); r != nil {
c.OK, c.Err = false, fmt.Sprintf("the check panicked: %v", r)
}
}()
if err := fn(); err != nil {
c.Err = err.Error()
return c
}
c.OK = true
return c
}
// Healthy reports whether every check passed.
func Healthy(checks []Check) bool {
for _, c := range checks {
if !c.OK {
return false
}
}
return true
}
func pingDB(ctx context.Context) error {
db := sdk.Runtime.GetDb()
if db == nil {
return errors.New("no database configured")
}
sqlDB, err := db.DB()
if err != nil {
return err
}
return sqlDB.PingContext(ctx)
}
// cacheProbePrefix names the probe's keys. The key itself is per probe, not
// fixed: two /ready requests arriving together - or two instances sharing one
// redis, which is the normal deployment - would otherwise overwrite each
// other's value between the write and the read and each conclude the cache was
// broken. A readiness probe that reports false negatives under load takes
// healthy instances out of the pool, which is worse than not probing.
const cacheProbePrefix = "go-admin:health:"
// cacheProbeTTL is short because these keys are write-once and never read
// again by anyone else; it only has to outlive the read that follows.
const cacheProbeTTL = 30
func probeCache() error {
adapter := sdk.Runtime.GetCacheAdapter()
if adapter == nil {
return errors.New("no cache configured")
}
suffix := make([]byte, 8)
if _, err := rand.Read(suffix); err != nil {
return fmt.Errorf("could not build a probe key: %w", err)
}
key := cacheProbePrefix + hex.EncodeToString(suffix)
// Written and read back rather than only read: a cache that answers "miss"
// for every key - a client pointed at the wrong server - is
// indistinguishable from a healthy one on a read alone.
want := time.Now().Format(time.RFC3339Nano)
if err := adapter.Set(key, want, cacheProbeTTL); err != nil {
return err
}
// Best effort, and its error is deliberately dropped: the verdict is
// already decided by the read below, and a cache that cannot delete a key
// it just wrote is not a reason to refuse traffic. The TTL is the real
// cleanup.
defer func() { _ = adapter.Del(key) }()
got, err := adapter.Get(key)
if err != nil {
return err
}
if got != want {
return errors.New("the cache returned a different value than was written")
}
return nil
}
+241
View File
@@ -0,0 +1,241 @@
package health
import (
"context"
"errors"
"sync"
"testing"
"time"
"github.com/go-admin-team/go-admin-core/v2/sdk"
"github.com/go-admin-team/go-admin-core/v2/sdk/runtime"
corestorage "github.com/go-admin-team/go-admin-core/v2/storage"
)
func freshRuntime(t *testing.T) {
t.Helper()
previous := sdk.Runtime
t.Cleanup(func() { sdk.Runtime = previous })
sdk.Runtime = runtime.NewConfig()
}
// fakeCache answers whatever the test needs it to.
type fakeCache struct {
mu sync.Mutex
setErr error
getErr error
getBack string // returned instead of what was written, when non-empty
stored map[string]string
// oneSlot makes the cache keep a single value however many keys are
// written, which is what a shared probe key turns any cache into.
oneSlot bool
slot string
// setBarrier, when set, holds every writer until all of them have written.
// Without it the probes are short enough that the scheduler usually runs
// them one after another, and a shared key survives by luck rather than by
// design - which would leave the test below asserting nothing.
setBarrier *barrier
}
// barrier releases every waiter once n of them have arrived.
type barrier struct {
n int
mu sync.Mutex
got int
ch chan struct{}
}
func newBarrier(n int) *barrier { return &barrier{n: n, ch: make(chan struct{})} }
func (b *barrier) wait() {
b.mu.Lock()
b.got++
if b.got == b.n {
close(b.ch)
}
b.mu.Unlock()
<-b.ch
}
func (c *fakeCache) String() string { return "fake" }
func (c *fakeCache) Set(key string, val interface{}, _ int) error {
c.mu.Lock()
defer c.mu.Unlock()
if c.setErr != nil {
return c.setErr
}
v, _ := val.(string)
if c.oneSlot {
c.slot = v
return nil
}
if c.stored == nil {
c.stored = map[string]string{}
}
c.stored[key] = v
c.mu.Unlock()
if c.setBarrier != nil {
// Outside the lock on purpose: waiting while holding it would deadlock
// every other writer before the barrier could fill.
c.setBarrier.wait()
}
c.mu.Lock()
return nil
}
func (c *fakeCache) Get(key string) (string, error) {
c.mu.Lock()
defer c.mu.Unlock()
if c.getErr != nil {
return "", c.getErr
}
if c.getBack != "" {
return c.getBack, nil
}
if c.oneSlot {
return c.slot, nil
}
return c.stored[key], nil
}
func (c *fakeCache) Del(key string) error {
c.mu.Lock()
defer c.mu.Unlock()
delete(c.stored, key)
return nil
}
func (c *fakeCache) HashGet(_, _ string) (string, error) { return "", nil }
func (c *fakeCache) HashDel(_, _ string) error { return nil }
func (c *fakeCache) Increase(string) error { return nil }
func (c *fakeCache) Decrease(string) error { return nil }
func (c *fakeCache) Expire(string, time.Duration) error { return nil }
var _ corestorage.AdapterCache = (*fakeCache)(nil)
func named(checks []Check, name string) Check {
for _, c := range checks {
if c.Name == name {
return c
}
}
return Check{Name: name, Err: "check not reported at all"}
}
// A cache that accepts writes and answers every read with a different value is
// the failure this probe exists for - a client pointed at the wrong server, or
// one that silently drops everything. A read alone cannot tell that apart from
// a healthy cache with a cold key, which is why the probe writes first.
func TestCacheProbeFailsWhenTheValueDoesNotComeBack(t *testing.T) {
freshRuntime(t)
sdk.Runtime.SetCacheAdapter(&fakeCache{getBack: "something else"})
got := named(Ready(context.Background()), "cache")
if got.OK {
t.Error("the cache check passed although the value written was not the value read back")
}
if got.Err == "" {
t.Error("the failing check reported no reason")
}
}
func TestCacheProbePassesWhenTheValueComesBack(t *testing.T) {
freshRuntime(t)
sdk.Runtime.SetCacheAdapter(&fakeCache{})
if got := named(Ready(context.Background()), "cache"); !got.OK {
t.Errorf("the cache check failed for a cache that works: %s", got.Err)
}
}
func TestCacheProbeReportsAWriteFailure(t *testing.T) {
freshRuntime(t)
sdk.Runtime.SetCacheAdapter(&fakeCache{setErr: errors.New("connection refused")})
got := named(Ready(context.Background()), "cache")
if got.OK {
t.Error("the cache check passed although the write failed")
}
}
// Nothing configured is the case that used to take the process down rather
// than answer. GetCacheAdapter builds a wrapper around whatever is configured
// and returns it even when nothing is, so the value is not nil, the cache
// inside it is, and Set dereferences it - a probe that panics is the worst
// possible answer to "are you well".
//
// Every check has to be reported, passing or not. A probe that omits what it
// could not reach reads as a shorter list of healthy things.
func TestEveryDependencyIsReportedEvenWithNothingConfigured(t *testing.T) {
freshRuntime(t)
checks := Ready(context.Background())
for _, name := range []string{"database", "cache"} {
c := named(checks, name)
if c.Err == "check not reported at all" {
t.Errorf("%s was not reported", name)
}
if c.OK {
t.Errorf("%s passed with nothing configured", name)
}
}
if Healthy(checks) {
t.Error("Healthy said yes for a process with no database and no cache")
}
}
// Draining is what makes the shutdown graceful from the outside: it has to be
// observable before the server stops accepting, or the load balancer learns
// about the shutdown by having its connections cut.
func TestDrainingIsObservableOnceItBegins(t *testing.T) {
previous := draining.Load()
t.Cleanup(func() { draining.Store(previous) })
draining.Store(false)
if Draining() {
t.Fatal("Draining reported true before shutdown began")
}
BeginDraining()
if !Draining() {
t.Error("Draining still reported false after BeginDraining")
}
}
// Two probes at once must both pass. With one fixed key they overwrite each
// other's value between the write and the read, and a readiness probe that
// reports false negatives under load takes healthy instances out of the pool -
// which is worse than not probing at all.
func TestConcurrentProbesDoNotOverwriteEachOther(t *testing.T) {
freshRuntime(t)
const probes = 16
sdk.Runtime.SetCacheAdapter(&fakeCache{setBarrier: newBarrier(probes)})
var wg sync.WaitGroup
failures := make(chan string, probes)
for i := 0; i < probes; i++ {
wg.Add(1)
go func() {
defer wg.Done()
if c := named(Ready(context.Background()), "cache"); !c.OK {
failures <- c.Err
}
}()
}
wg.Wait()
close(failures)
var n int
var first string
for err := range failures {
if n == 0 {
first = err
}
n++
}
if n > 0 {
t.Errorf("%d of %d concurrent probes called a healthy cache broken; first: %s", n, probes, first)
}
}
+1
View File
@@ -34,6 +34,7 @@ var CasbinExclude = []UrlInfo{
{Url: "/api/v1/user/pwd", Method: "PUT"},
{Url: "/api/v1/metrics", Method: "GET"},
{Url: "/api/v1/health", Method: "GET"},
{Url: "/api/v1/ready", Method: "GET"},
{Url: "/", Method: "GET"},
{Url: "/api/v1/server-monitor", Method: "GET"},
{Url: "/api/v1/public/uploadFile", Method: "POST"},
+64 -5
View File
@@ -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,75 @@ 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()
queueAdapter, err := config.QueueConfig.Setup()
if err != nil {
log.Fatalf("queue setup error, %s\n", err.Error())
}
previous := installed
sdk.Runtime.SetQueueAdapter(queueAdapter)
go queueAdapter.Run()
installed = queueAdapter
installedGen++
// The previous adapter goes down after the new one is installed, not
// before. Shutdown waits for its consumers to deliver what it still holds,
// and for that whole wait the runtime would otherwise be handing producers
// a queue that has stopped accepting: every Append in the window comes back
// ErrQueueClosed, and both call sites in common/middleware log it. Swapping
// first leaves no such window - a producer gets the new queue or the old
// one, and both work.
//
// 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 previous != nil {
previous.Shutdown()
}
// 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.
}
+130
View File
@@ -0,0 +1,130 @@
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 != "" {
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
// 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)
}
}
+161
View File
@@ -0,0 +1,161 @@
package storage
import (
"errors"
"sync"
"sync/atomic"
"testing"
"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"
corestorage "github.com/go-admin-team/go-admin-core/v2/storage"
"github.com/go-admin-team/go-admin-core/v2/storage/queue"
)
// sampleSize is how many publishes have to land inside the reload before the
// measurement is taken. Waiting on the count rather than on wall clock keeps
// the window the test covers the same on a loaded runner as on an idle one.
const sampleSize = 200
func swapMsg() corestorage.Messager {
m := new(queue.Message)
m.SetStream("t")
m.SetValues(map[string]interface{}{"a": "b"})
return m
}
// A reload must never leave producers holding a queue that has stopped
// accepting.
//
// Shutdown waits for its consumers to deliver what the queue still holds. Taking
// the old adapter down before installing the new one meant the runtime pointed
// at a closed queue for that entire wait: every Append in the window came back
// ErrQueueClosed, and both call sites in common/middleware log it at error
// level. Installing first leaves no window - a producer gets the new queue or
// the old one, and both accept.
//
// The difference is only visible during that wait, which is why the test holds
// a consumer rather than checking the state after Setup has returned: by then
// the two orders look identical.
//
// One refusal survives the fix and is not something this ordering can reach.
// GetQueuePrefix hands back a wrapper that captured the adapter, so a producer
// that fetched before the swap and appends after Shutdown has begun is still
// holding the old one. That window is one call wide and closing it means
// resolving the adapter inside Append, which is core's to change. What the
// ordering removes is the sustained window: every producer that fetches during
// the wait. The test publishes from a single goroutine, so at most one of its
// calls can straddle the swap - which is what makes "more than one" the line
// between the two orders rather than a tolerance.
func TestAReloadNeverPointsProducersAtAClosedQueue(t *testing.T) {
prevQ, prevC := config.QueueConfig, config.CacheConfig
prevRuntime := sdk.Runtime
prevInstalled, prevGen := installed, installedGen
t.Cleanup(func() {
config.QueueConfig, config.CacheConfig = prevQ, prevC
sdk.Runtime = prevRuntime
queueMu.Lock()
installed, installedGen = prevInstalled, prevGen
queueMu.Unlock()
})
sdk.Runtime = runtime.NewConfig()
config.CacheConfig = &config.Cache{Memory: struct{}{}}
// Sized so the buffer cannot fill while the consumer is held: a full queue
// returns an error of its own, and this test needs every error other than
// ErrQueueClosed to mean something it does not model has happened.
config.QueueConfig = &config.Queue{Memory: &config.QueueMemory{PoolSize: 4096}}
Setup()
// A consumer that will not finish until this test lets it, so the reload's
// Shutdown has something to wait for.
release := make(chan struct{})
consuming := make(chan struct{})
var picked sync.Once
first := sdk.Runtime.GetQueuePrefix("")
first.Register("t", func(corestorage.Messager) error {
picked.Do(func() { close(consuming) })
<-release
return nil
})
go first.Run()
for i := 0; i < 4; i++ {
if err := first.Append(swapMsg()); err != nil {
t.Fatalf("seed append %d: %v", i, err)
}
}
select {
case <-consuming:
case <-time.After(10 * time.Second):
t.Fatal("the consumer never picked a message up, so the reload has nothing to wait for")
}
reloaded := make(chan struct{})
go func() { Setup(); close(reloaded) }()
// Publish continuously while the reload is in progress.
var refused atomic.Int64
var attempts atomic.Int64
unexpected := make(chan error, 1)
stop := make(chan struct{})
// publishing is closed by the producer on its way out. The test joins on it
// before returning: t.Cleanup restores sdk.Runtime, and a producer still in
// flight would be reading the variable that restore writes.
publishing := make(chan struct{})
go func() {
defer close(publishing)
for {
select {
case <-stop:
return
default:
}
attempts.Add(1)
err := sdk.Runtime.GetQueuePrefix("").Append(swapMsg())
switch {
case err == nil:
case errors.Is(err, corestorage.ErrQueueClosed):
refused.Add(1)
default:
// Kept rather than counted: an Append refused for some other
// reason would otherwise leave refused at zero and the test
// green while nothing was reaching a queue at all.
select {
case unexpected <- err:
default:
}
}
time.Sleep(time.Millisecond)
}
}()
deadline := time.After(30 * time.Second)
for attempts.Load() < sampleSize {
select {
case <-deadline:
t.Fatalf("only %d publishes landed inside the reload; the window was never sampled", attempts.Load())
case <-time.After(time.Millisecond):
}
}
close(release)
select {
case <-reloaded:
case <-time.After(30 * time.Second):
t.Fatal("the reload never finished")
}
close(stop)
<-publishing
select {
case err := <-unexpected:
t.Fatalf("a publish failed for a reason this test does not model: %v", err)
default:
}
if n := refused.Load(); n > 1 {
t.Errorf("%d of %d publishes during the reload were refused: producers were pointed at the closed queue",
n, attempts.Load())
}
}
+53
View File
@@ -0,0 +1,53 @@
package storage
import (
"testing"
"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"
)
// Issue #892: a configuration reload replaces the queue adapter and the
// consumers registered against the previous one are attached to a queue nobody
// publishes to any more.
//
// The fix has two halves. attachQueueConsumers gives a new queue its own
// consumers and the same queue none, which cmd/api covers against a queue the
// test controls. This is the other half: that a reload actually produces a new
// queue for it to notice. Setup is what config re-runs on every change, so
// calling it twice is what a reload does to this package.
func TestSetupBumpsTheQueueGenerationOnEveryReload(t *testing.T) {
// Setup writes the process-wide sdk.Runtime - the cache and queue adapters -
// and this package's own record of what it installed. Restoring all of it
// keeps the test from deciding what a later test in this binary sees,
// which is the same isolation cmd/api's freshRuntime provides.
prevQ, prevC := config.QueueConfig, config.CacheConfig
prevRuntime := sdk.Runtime
prevInstalled, prevGen := installed, installedGen
t.Cleanup(func() {
config.QueueConfig, config.CacheConfig = prevQ, prevC
sdk.Runtime = prevRuntime
queueMu.Lock()
installed, installedGen = prevInstalled, prevGen
queueMu.Unlock()
})
sdk.Runtime = runtime.NewConfig()
config.CacheConfig = &config.Cache{Memory: struct{}{}}
config.QueueConfig = &config.Queue{Memory: &config.QueueMemory{PoolSize: 10}}
before := QueueGeneration()
Setup()
first := QueueGeneration()
Setup()
second := QueueGeneration()
t.Logf("before=%d first=%d second=%d", before, first, second)
if first == before {
t.Fatal("the first Setup did not install a queue")
}
if second == first {
t.Fatal("a second Setup - which is what a configuration reload does - did not install a new one")
}
}
+86 -1
View File
@@ -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,
钩子在它和预算之间二选一。
---
## 安全边界:装一个应用等于信任它
**这一层划不出安全边界,本文不假装划得出。**
+1 -1
View File
@@ -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.7.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
+4
View File
@@ -147,6 +147,10 @@ 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-admin-team/go-admin-core/v2 v2.7.0 h1:1qV0/5iFBvkE3BRtm4ip0v0QYG9Fgx4UtOTd8zkQT9c=
github.com/go-admin-team/go-admin-core/v2 v2.7.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=