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.

app/other/router/monitor.go

@@ -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})
+	})
+
+}

cmd/api/server.go

@@ -30,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"
@@ -217,6 +218,12 @@ func run() error {
 	// 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 {

common/health/health.go

@@ -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
+}

common/health/health_test.go

@@ -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)
+	}
+}

common/middleware/settings.go

@@ -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"},

common/storage/initialize.go

@@ -71,23 +71,33 @@ func setupQueue() {
 	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)
+	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 installed != nil {
-		installed.Shutdown()
+	if previous != nil {
+		previous.Shutdown()
 	}
 
-	queueAdapter, err := config.QueueConfig.Setup()
-	if err != nil {
-		log.Fatalf("queue setup error, %s\n", err.Error())
-	}
-	sdk.Runtime.SetQueueAdapter(queueAdapter)
-	installed = queueAdapter
-	installedGen++
-
 	// Deliberately not started here. Run has to come after the consumers have
 	// registered: the contract implementations refuse a registration once the
 	// queue is running (storage.ErrQueueAlreadyStarted), and the legacy

common/storage/queue_swap_test.go

@@ -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())
+	}
+}

common/storage/setup_gen_test.go

@@ -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")
+	}
+}

go.mod

@@ -11,7 +11,7 @@ require (
 	github.com/casbin/casbin/v3 v3.8.1
 	github.com/gin-gonic/gin v1.12.0
 	github.com/glebarez/sqlite v1.11.0
-	github.com/go-admin-team/go-admin-core/v2 v2.6.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

go.sum

@@ -149,6 +149,8 @@ github.com/go-admin-team/go-admin-core/v2 v2.5.0 h1:aD1SALklBxizGB9u8cOgm4OT8z65
 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=