From 46f4092b43dadd4a682a3e33d3af3db21c9f0c09 Mon Sep 17 00:00:00 2001 From: zhangwenjian Date: Sun, 6 Sep 2026 17:37:01 +0800 Subject: [PATCH 1/6] =?UTF-8?q?build=F0=9F=94=A7:=20require=20go-admin-cor?= =?UTF-8?q?e=20v2.7.0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- go.mod | 2 +- go.sum | 2 ++ 2 files changed, 3 insertions(+), 1 deletion(-) diff --git a/go.mod b/go.mod index 899f5147..c0a01c6a 100644 --- a/go.mod +++ b/go.mod @@ -11,7 +11,7 @@ require ( github.com/casbin/casbin/v3 v3.8.1 github.com/gin-gonic/gin v1.12.0 github.com/glebarez/sqlite v1.11.0 - github.com/go-admin-team/go-admin-core/v2 v2.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 diff --git a/go.sum b/go.sum index a79c5d22..de41fc4b 100644 --- a/go.sum +++ b/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= From 46e793972c6a96188642baba2387a747d43a383b Mon Sep 17 00:00:00 2001 From: zhangwenjian Date: Sun, 6 Sep 2026 17:37:01 +0800 Subject: [PATCH 2/6] =?UTF-8?q?fix=F0=9F=90=9B:=20install=20the=20new=20qu?= =?UTF-8?q?eue=20before=20shutting=20the=20old=20one=20down?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- common/storage/initialize.go | 30 ++++++--- common/storage/queue_swap_test.go | 108 ++++++++++++++++++++++++++++++ 2 files changed, 128 insertions(+), 10 deletions(-) create mode 100644 common/storage/queue_swap_test.go diff --git a/common/storage/initialize.go b/common/storage/initialize.go index b1f552cb..305fc780 100644 --- a/common/storage/initialize.go +++ b/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 diff --git a/common/storage/queue_swap_test.go b/common/storage/queue_swap_test.go new file mode 100644 index 00000000..cbe44f58 --- /dev/null +++ b/common/storage/queue_swap_test.go @@ -0,0 +1,108 @@ +package storage + +import ( + "errors" + "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" +) + +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. +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{}{}} + config.QueueConfig = &config.Queue{Memory: &config.QueueMemory{PoolSize: 64}} + + 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{}) + first := sdk.Runtime.GetQueuePrefix("") + first.Register("t", func(corestorage.Messager) error { + <-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) + } + } + time.Sleep(100 * time.Millisecond) // let the consumer pick one up and block + + 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 + stop := make(chan struct{}) + go func() { + for { + select { + case <-stop: + return + default: + } + attempts.Add(1) + if err := sdk.Runtime.GetQueuePrefix("").Append(swapMsg()); errors.Is(err, corestorage.ErrQueueClosed) { + refused.Add(1) + } + time.Sleep(time.Millisecond) + } + }() + + time.Sleep(200 * time.Millisecond) // the reload is now inside Shutdown's wait + close(release) + + select { + case <-reloaded: + case <-time.After(10 * time.Second): + t.Fatal("the reload never finished") + } + close(stop) + + if attempts.Load() == 0 { + t.Fatal("nothing was published during the reload; the test proves nothing") + } + if n := refused.Load(); n > 0 { + t.Errorf("%d of %d publishes during the reload were refused: producers were pointed at the closed queue", + n, attempts.Load()) + } +} From d5de79f75ba4fc6c285cf53c54a9d055faea50ad Mon Sep 17 00:00:00 2001 From: zhangwenjian Date: Sun, 6 Sep 2026 19:00:10 +0800 Subject: [PATCH 3/6] =?UTF-8?q?test=E2=9C=85:=20join=20the=20producer=20be?= =?UTF-8?q?fore=20the=20test=20returns?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- common/storage/queue_swap_test.go | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/common/storage/queue_swap_test.go b/common/storage/queue_swap_test.go index cbe44f58..6d253363 100644 --- a/common/storage/queue_swap_test.go +++ b/common/storage/queue_swap_test.go @@ -73,7 +73,12 @@ func TestAReloadNeverPointsProducersAtAClosedQueue(t *testing.T) { var refused atomic.Int64 var attempts atomic.Int64 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: @@ -97,6 +102,7 @@ func TestAReloadNeverPointsProducersAtAClosedQueue(t *testing.T) { t.Fatal("the reload never finished") } close(stop) + <-publishing if attempts.Load() == 0 { t.Fatal("nothing was published during the reload; the test proves nothing") From 6138d2d74cdd318c84ce00b0f5c6f6786dd579d5 Mon Sep 17 00:00:00 2001 From: zhangwenjian Date: Sun, 6 Sep 2026 19:00:20 +0800 Subject: [PATCH 4/6] =?UTF-8?q?test=E2=9C=85:=20assert=20the=20bound=20ins?= =?UTF-8?q?talling=20first=20actually=20gives?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- common/storage/queue_swap_test.go | 12 +++++++++++- 1 file changed, 11 insertions(+), 1 deletion(-) diff --git a/common/storage/queue_swap_test.go b/common/storage/queue_swap_test.go index 6d253363..60c64f0a 100644 --- a/common/storage/queue_swap_test.go +++ b/common/storage/queue_swap_test.go @@ -33,6 +33,16 @@ func swapMsg() corestorage.Messager { // 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 @@ -107,7 +117,7 @@ func TestAReloadNeverPointsProducersAtAClosedQueue(t *testing.T) { if attempts.Load() == 0 { t.Fatal("nothing was published during the reload; the test proves nothing") } - if n := refused.Load(); n > 0 { + 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()) } From 3c3d94ca76d3362ff7c29eb025dafe686ec14aa2 Mon Sep 17 00:00:00 2001 From: zhangwenjian Date: Sun, 6 Sep 2026 19:00:25 +0800 Subject: [PATCH 5/6] =?UTF-8?q?test=E2=9C=85:=20wait=20for=20the=20sample?= =?UTF-8?q?=20rather=20than=20for=20the=20clock?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- common/storage/queue_swap_test.go | 29 +++++++++++++++++++++++------ 1 file changed, 23 insertions(+), 6 deletions(-) diff --git a/common/storage/queue_swap_test.go b/common/storage/queue_swap_test.go index 60c64f0a..ff8d4abe 100644 --- a/common/storage/queue_swap_test.go +++ b/common/storage/queue_swap_test.go @@ -2,6 +2,7 @@ package storage import ( "errors" + "sync" "sync/atomic" "testing" "time" @@ -13,6 +14,11 @@ import ( "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") @@ -63,8 +69,11 @@ func TestAReloadNeverPointsProducersAtAClosedQueue(t *testing.T) { // 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 }) @@ -74,7 +83,11 @@ func TestAReloadNeverPointsProducersAtAClosedQueue(t *testing.T) { t.Fatalf("seed append %d: %v", i, err) } } - time.Sleep(100 * time.Millisecond) // let the consumer pick one up and block + 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) }() @@ -103,20 +116,24 @@ func TestAReloadNeverPointsProducersAtAClosedQueue(t *testing.T) { } }() - time.Sleep(200 * time.Millisecond) // the reload is now inside Shutdown's wait + 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(10 * time.Second): + case <-time.After(30 * time.Second): t.Fatal("the reload never finished") } close(stop) <-publishing - if attempts.Load() == 0 { - t.Fatal("nothing was published during the reload; the test proves nothing") - } 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()) From 211ae85a4edf005a2056db9d02f8419e1835e53b Mon Sep 17 00:00:00 2001 From: zhangwenjian Date: Sun, 6 Sep 2026 19:00:30 +0800 Subject: [PATCH 6/6] =?UTF-8?q?test=E2=9C=85:=20fail=20on=20an=20Append=20?= =?UTF-8?q?error=20this=20test=20does=20not=20model?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- common/storage/queue_swap_test.go | 24 ++++++++++++++++++++++-- 1 file changed, 22 insertions(+), 2 deletions(-) diff --git a/common/storage/queue_swap_test.go b/common/storage/queue_swap_test.go index ff8d4abe..14d29cad 100644 --- a/common/storage/queue_swap_test.go +++ b/common/storage/queue_swap_test.go @@ -62,7 +62,10 @@ func TestAReloadNeverPointsProducersAtAClosedQueue(t *testing.T) { }) sdk.Runtime = runtime.NewConfig() config.CacheConfig = &config.Cache{Memory: struct{}{}} - config.QueueConfig = &config.Queue{Memory: &config.QueueMemory{PoolSize: 64}} + // 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() @@ -95,6 +98,7 @@ func TestAReloadNeverPointsProducersAtAClosedQueue(t *testing.T) { // 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 @@ -109,8 +113,19 @@ func TestAReloadNeverPointsProducersAtAClosedQueue(t *testing.T) { default: } attempts.Add(1) - if err := sdk.Runtime.GetQueuePrefix("").Append(swapMsg()); errors.Is(err, corestorage.ErrQueueClosed) { + 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) } @@ -134,6 +149,11 @@ func TestAReloadNeverPointsProducersAtAClosedQueue(t *testing.T) { 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())