From 32bde537c0bd046193f3d426b8de376a1cd6d1b8 Mon Sep 17 00:00:00 2001 From: 3metaJun <251347867+3metaJun@users.noreply.github.com> Date: Thu, 3 Sep 2026 17:21:59 +0800 Subject: [PATCH] feat: hot-reload provision releases instead of freezing for a restart ApplyMetaproxyProvision used to freeze the periodic option and channel syncs after committing and hand a RestartRequired flag back to the host, which then put nginx into maintenance mode and restarted the process. Every price, model or channel change was a user-visible 503 window. The transaction now reloads the in-process state itself, still under the provision runtime write lock: options and ratios first, then the channel cache, so a model is never routable before its ratio exists. Periodic syncs take the read lock, so they cannot interleave with a reload. An idempotent retry of the already-active digest skips the reload. Co-Authored-By: Claude Fable 5.1 --- controller/metaproxy_provision.go | 9 ++- controller/metaproxy_provision_test.go | 77 +++++++++++++++++++++++ model/channel_cache.go | 7 +-- model/metaproxy_provision.go | 39 +++++++----- model/metaproxy_provision_test.go | 87 ++++++++++++++++++++++---- model/option.go | 7 +-- 6 files changed, 181 insertions(+), 45 deletions(-) diff --git a/controller/metaproxy_provision.go b/controller/metaproxy_provision.go index 54ebf8f82937..4aa1a773e1c1 100644 --- a/controller/metaproxy_provision.go +++ b/controller/metaproxy_provision.go @@ -357,11 +357,10 @@ func ApplyMetaproxyProvision(c *gin.Context) { "success": true, "message": "", "data": gin.H{ - "revision": request.Revision, - "digest": request.Digest, - "previous_digest": result.PreviousDigest, - "already_applied": result.AlreadyApplied, - "restart_required": result.RestartRequired, + "revision": request.Revision, + "digest": request.Digest, + "previous_digest": result.PreviousDigest, + "already_applied": result.AlreadyApplied, }, }) } diff --git a/controller/metaproxy_provision_test.go b/controller/metaproxy_provision_test.go index adfd0f6af7f4..b01772ac7231 100644 --- a/controller/metaproxy_provision_test.go +++ b/controller/metaproxy_provision_test.go @@ -5,11 +5,15 @@ import ( "encoding/json" "net/http" "net/http/httptest" + "sort" "testing" "github.com/QuantumNous/new-api/common" + "github.com/QuantumNous/new-api/model" "github.com/gin-gonic/gin" + "github.com/glebarez/sqlite" "github.com/stretchr/testify/require" + "gorm.io/gorm" ) func validMetaproxyProvisionRequest() metaproxyProvisionRequest { @@ -193,3 +197,76 @@ func TestApplyMetaproxyProvisionRequiresMemoryCache(t *testing.T) { require.Equal(t, http.StatusPreconditionFailed, recorder.Code) require.Contains(t, recorder.Body.String(), "MEMORY_CACHE_ENABLED=true") } + +func TestApplyMetaproxyProvisionSuccessResponseShape(t *testing.T) { + db, err := gorm.Open(sqlite.Open(":memory:"), &gorm.Config{}) + require.NoError(t, err) + require.NoError(t, db.AutoMigrate(&model.Channel{}, &model.Ability{}, &model.Option{}, &model.Log{})) + request := validMetaproxyProvisionRequest() + require.NoError(t, db.Create(&[]model.Option{ + {Key: model.MetaproxyProvisionDigestOption, Value: request.Digest}, + {Key: model.MetaproxyProvisionRevisionOption, Value: request.Revision}, + }).Error) + + previousDB, previousLogDB := model.DB, model.LOG_DB + previousMemoryCacheEnabled := common.MemoryCacheEnabled + previousRedisEnabled := common.RedisEnabled + common.RedisEnabled = false + // Pre-activate the digest in memory so the apply skips the in-process + // reload: the reload steps are model-internal and must not run here. + common.OptionMapRWMutex.Lock() + previousOptions := common.OptionMap + common.OptionMap = map[string]string{ + model.MetaproxyProvisionDigestOption: request.Digest, + model.MetaproxyProvisionRevisionOption: request.Revision, + } + common.OptionMapRWMutex.Unlock() + model.DB, model.LOG_DB = db, db + common.MemoryCacheEnabled = true + t.Cleanup(func() { + model.DB, model.LOG_DB = previousDB, previousLogDB + common.MemoryCacheEnabled = previousMemoryCacheEnabled + common.RedisEnabled = previousRedisEnabled + common.OptionMapRWMutex.Lock() + common.OptionMap = previousOptions + common.OptionMapRWMutex.Unlock() + }) + + body, err := json.Marshal(request) + require.NoError(t, err) + recorder := httptest.NewRecorder() + context, _ := gin.CreateTestContext(recorder) + context.Request = httptest.NewRequest(http.MethodPost, "/api/metaproxy/provision", bytes.NewReader(body)) + context.Request.Header.Set("Idempotency-Key", request.Digest) + context.Request.Header.Set("If-Match", request.Digest) + + ApplyMetaproxyProvision(context) + + require.Equal(t, http.StatusOK, recorder.Code) + var payload struct { + Success bool `json:"success"` + Data struct { + Revision string `json:"revision"` + Digest string `json:"digest"` + PreviousDigest string `json:"previous_digest"` + AlreadyApplied bool `json:"already_applied"` + } `json:"data"` + } + require.NoError(t, json.Unmarshal(recorder.Body.Bytes(), &payload)) + require.True(t, payload.Success) + require.Equal(t, request.Revision, payload.Data.Revision) + require.Equal(t, request.Digest, payload.Data.Digest) + require.Equal(t, request.Digest, payload.Data.PreviousDigest) + require.False(t, payload.Data.AlreadyApplied) + + var raw map[string]any + require.NoError(t, json.Unmarshal(recorder.Body.Bytes(), &raw)) + data, ok := raw["data"].(map[string]any) + require.True(t, ok) + keys := make([]string, 0, len(data)) + for key := range data { + keys = append(keys, key) + } + sort.Strings(keys) + require.Equal(t, []string{"already_applied", "digest", "previous_digest", "revision"}, keys) +} diff --git a/model/channel_cache.go b/model/channel_cache.go index 310037cc4d9e..43a7ff627d20 100644 --- a/model/channel_cache.go +++ b/model/channel_cache.go @@ -106,13 +106,10 @@ func InitChannelCache() { func SyncChannelCache(frequency int) { for { time.Sleep(time.Duration(frequency) * time.Second) - if !RunMetaproxyProvisionSyncIfReady(func() { + runProvisionSync(func() { common.SysLog("syncing channels from database") InitChannelCache() - }) { - common.SysLog("skipping channel sync while a metaproxy provision restart is pending") - continue - } + }) } } diff --git a/model/metaproxy_provision.go b/model/metaproxy_provision.go index ed6a9785c7dc..609046097132 100644 --- a/model/metaproxy_provision.go +++ b/model/metaproxy_provision.go @@ -4,7 +4,6 @@ import ( "errors" "fmt" "sync" - "sync/atomic" "github.com/QuantumNous/new-api/common" "gorm.io/gorm" @@ -23,7 +22,14 @@ var ( ErrMetaproxyProvisionRequiresMemoryCache = errors.New("metaproxy provision requires MEMORY_CACHE_ENABLED=true") metaproxyProvisionLock sync.Mutex metaproxyProvisionRuntimeLock sync.RWMutex - provisionRuntimeFrozen atomic.Bool + + // reloadProvisionOptions and reloadProvisionChannels are the in-process + // reload steps run after a provision apply commits. They are package-level + // variables so tests can stub and record them. Ordering is load-bearing: + // options/ratios must be reloaded before the channel cache so a model is + // never routable before its ratio exists. + reloadProvisionOptions = loadOptionsFromDatabase + reloadProvisionChannels = InitChannelCache ) type MetaproxyProvisionChannel struct { @@ -71,23 +77,17 @@ type MetaproxyProvisionConfig struct { } type MetaproxyProvisionResult struct { - AlreadyApplied bool - RestartRequired bool - PreviousDigest string -} - -func IsMetaproxyProvisionRuntimeFrozen() bool { - return provisionRuntimeFrozen.Load() + AlreadyApplied bool + PreviousDigest string } -func RunMetaproxyProvisionSyncIfReady(syncFn func()) bool { +// runProvisionSync runs a periodic sync under the provision runtime read lock +// so syncs can never interleave with an in-flight provision apply reload +// (which holds the write lock). +func runProvisionSync(syncFn func()) { metaproxyProvisionRuntimeLock.RLock() defer metaproxyProvisionRuntimeLock.RUnlock() - if provisionRuntimeFrozen.Load() { - return false - } syncFn() - return true } func activeMetaproxyProvisionDigest() string { @@ -329,9 +329,14 @@ func ApplyMetaproxyProvision( return MetaproxyProvisionResult{}, err } - result.RestartRequired = activeMetaproxyProvisionDigest() != config.Digest - if result.RestartRequired { - provisionRuntimeFrozen.Store(true) + // The transaction committed; hot-reload the in-process state so no restart + // is required. Only reload when the active in-memory digest differs from + // the applied one, so an idempotent retry of the already-active digest is a + // no-op. Still holding metaproxyProvisionRuntimeLock for writing, so + // periodic syncs cannot interleave with the reload. + if activeMetaproxyProvisionDigest() != config.Digest { + reloadProvisionOptions() + reloadProvisionChannels() } return result, nil } diff --git a/model/metaproxy_provision_test.go b/model/metaproxy_provision_test.go index 610959dc6f0a..9e3d5c3e72e1 100644 --- a/model/metaproxy_provision_test.go +++ b/model/metaproxy_provision_test.go @@ -22,7 +22,6 @@ func setupMetaproxyProvisionTestDB(t *testing.T) *gorm.DB { previousMemoryCacheEnabled := common.MemoryCacheEnabled DB = db common.MemoryCacheEnabled = true - provisionRuntimeFrozen.Store(false) common.OptionMapRWMutex.Lock() previousOptions := common.OptionMap common.OptionMap = map[string]string{ @@ -33,7 +32,6 @@ func setupMetaproxyProvisionTestDB(t *testing.T) *gorm.DB { t.Cleanup(func() { DB = previousDB common.MemoryCacheEnabled = previousMemoryCacheEnabled - provisionRuntimeFrozen.Store(false) common.OptionMapRWMutex.Lock() common.OptionMap = previousOptions common.OptionMapRWMutex.Unlock() @@ -41,6 +39,17 @@ func setupMetaproxyProvisionTestDB(t *testing.T) *gorm.DB { return db } +// stubProvisionReload replaces the hot-reload steps for the duration of one +// test and restores them afterwards. +func stubProvisionReload(t *testing.T, reloadOptions func(), reloadChannels func()) { + t.Helper() + originalOptions, originalChannels := reloadProvisionOptions, reloadProvisionChannels + reloadProvisionOptions, reloadProvisionChannels = reloadOptions, reloadChannels + t.Cleanup(func() { + reloadProvisionOptions, reloadProvisionChannels = originalOptions, originalChannels + }) +} + func provisionTestChannel(name, key, models string) MetaproxyProvisionChannel { return MetaproxyProvisionChannel{ Type: 1, @@ -115,12 +124,14 @@ func desiredProvisionConfig() MetaproxyProvisionConfig { func TestApplyMetaproxyProvisionReplacesManagedStateAtomically(t *testing.T) { db := setupMetaproxyProvisionTestDB(t) seedProvisionState(t, db) + config := desiredProvisionConfig() - result, err := ApplyMetaproxyProvision(desiredProvisionConfig(), "old-digest") + result, err := ApplyMetaproxyProvision(config, "old-digest") require.NoError(t, err) - require.False(t, result.AlreadyApplied) - require.True(t, result.RestartRequired) - require.True(t, IsMetaproxyProvisionRuntimeFrozen()) + require.Equal(t, MetaproxyProvisionResult{ + AlreadyApplied: false, + PreviousDigest: "old-digest", + }, result) var channels []Channel require.NoError(t, db.Where("tag = ?", MetaproxyProvisionManagedTag).Find(&channels).Error) @@ -141,15 +152,60 @@ func TestApplyMetaproxyProvisionReplacesManagedStateAtomically(t *testing.T) { for _, option := range options { got[option.Key] = option.Value } - require.Equal(t, desiredProvisionConfig().Digest, got[MetaproxyProvisionDigestOption]) - require.Equal(t, desiredProvisionConfig().Revision, got[MetaproxyProvisionRevisionOption]) + require.Equal(t, config.Digest, got[MetaproxyProvisionDigestOption]) + require.Equal(t, config.Revision, got[MetaproxyProvisionRevisionOption]) require.Equal(t, `{"new-model":1.5}`, got["ModelRatio"]) require.Equal(t, `{"image-model":"tiered_expr"}`, got["billing_setting.billing_mode"]) require.Equal(t, `{"image-model":"tier(\"base\", 200000)"}`, got["billing_setting.billing_expr"]) + // The apply hot-reloads the in-process state: the new digest is active in + // memory and the new channel is routable without a restart. common.OptionMapRWMutex.RLock() - require.Equal(t, "old-digest", common.OptionMap[MetaproxyProvisionDigestOption]) + require.Equal(t, config.Digest, common.OptionMap[MetaproxyProvisionDigestOption]) + common.OptionMapRWMutex.RUnlock() + + channel, err := GetRandomSatisfiedChannel("standard", "new-model", 0, "") + require.NoError(t, err) + require.NotNil(t, channel) + require.Equal(t, "New upstream [new]", channel.Name) + require.Equal(t, "new-key", channel.Key) +} + +func TestApplyMetaproxyProvisionReloadsOptionsBeforeChannels(t *testing.T) { + db := setupMetaproxyProvisionTestDB(t) + seedProvisionState(t, db) + + var reloadSequence []string + stubProvisionReload(t, + func() { reloadSequence = append(reloadSequence, "options") }, + func() { reloadSequence = append(reloadSequence, "channels") }, + ) + + _, err := ApplyMetaproxyProvision(desiredProvisionConfig(), "old-digest") + require.NoError(t, err) + require.Equal(t, []string{"options", "channels"}, reloadSequence) +} + +func TestApplyMetaproxyProvisionIdempotentRetrySkipsReload(t *testing.T) { + db := setupMetaproxyProvisionTestDB(t) + seedProvisionState(t, db) + config := desiredProvisionConfig() + + first, err := ApplyMetaproxyProvision(config, "old-digest") + require.NoError(t, err) + require.False(t, first.AlreadyApplied) + common.OptionMapRWMutex.RLock() + require.Equal(t, config.Digest, common.OptionMap[MetaproxyProvisionDigestOption]) common.OptionMapRWMutex.RUnlock() + + optionsReloads, channelsReloads := 0, 0 + stubProvisionReload(t, func() { optionsReloads++ }, func() { channelsReloads++ }) + + second, err := ApplyMetaproxyProvision(config, config.Digest) + require.NoError(t, err) + require.True(t, second.AlreadyApplied) + require.Zero(t, optionsReloads) + require.Zero(t, channelsReloads) } func TestApplyMetaproxyProvisionConflictDoesNotWrite(t *testing.T) { @@ -158,7 +214,6 @@ func TestApplyMetaproxyProvisionConflictDoesNotWrite(t *testing.T) { _, err := ApplyMetaproxyProvision(desiredProvisionConfig(), "some-other-digest") require.ErrorIs(t, err, ErrMetaproxyProvisionConflict) - require.False(t, IsMetaproxyProvisionRuntimeFrozen()) var channel Channel require.NoError(t, db.First(&channel, old.Id).Error) @@ -166,6 +221,10 @@ func TestApplyMetaproxyProvisionConflictDoesNotWrite(t *testing.T) { var digest Option require.NoError(t, db.First(&digest, "key = ?", MetaproxyProvisionDigestOption).Error) require.Equal(t, "old-digest", digest.Value) + + common.OptionMapRWMutex.RLock() + require.Equal(t, "old-digest", common.OptionMap[MetaproxyProvisionDigestOption]) + common.OptionMapRWMutex.RUnlock() } func TestApplyMetaproxyProvisionRequiresMemoryCache(t *testing.T) { @@ -197,10 +256,9 @@ func TestApplyMetaproxyProvisionUpdatesInPlaceAndIsIdempotent(t *testing.T) { require.Equal(t, old.Status, updated.Status) require.Equal(t, "rotated-key", updated.Key) - second, err := ApplyMetaproxyProvision(config, "old-digest") + second, err := ApplyMetaproxyProvision(config, config.Digest) require.NoError(t, err) require.True(t, second.AlreadyApplied) - require.True(t, second.RestartRequired) var count int64 require.NoError(t, db.Model(&Channel{}).Where("tag = ?", MetaproxyProvisionManagedTag).Count(&count).Error) @@ -222,7 +280,6 @@ func TestApplyMetaproxyProvisionRollsBackEveryTableOnFailure(t *testing.T) { _, err := ApplyMetaproxyProvision(desiredProvisionConfig(), "old-digest") require.Error(t, err) require.False(t, errors.Is(err, ErrMetaproxyProvisionConflict)) - require.False(t, IsMetaproxyProvisionRuntimeFrozen()) var channels []Channel require.NoError(t, db.Where("tag = ?", MetaproxyProvisionManagedTag).Find(&channels).Error) @@ -236,4 +293,8 @@ func TestApplyMetaproxyProvisionRollsBackEveryTableOnFailure(t *testing.T) { var digest Option require.NoError(t, db.First(&digest, "key = ?", MetaproxyProvisionDigestOption).Error) require.Equal(t, "old-digest", digest.Value) + + common.OptionMapRWMutex.RLock() + require.Equal(t, "old-digest", common.OptionMap[MetaproxyProvisionDigestOption]) + common.OptionMapRWMutex.RUnlock() } diff --git a/model/option.go b/model/option.go index 358822479184..9a47cd8d2965 100644 --- a/model/option.go +++ b/model/option.go @@ -226,13 +226,10 @@ func loadOptionsFromDatabase() { func SyncOptions(frequency int) { for { time.Sleep(time.Duration(frequency) * time.Second) - if !RunMetaproxyProvisionSyncIfReady(func() { + runProvisionSync(func() { common.SysLog("syncing options from database") loadOptionsFromDatabase() - }) { - common.SysLog("skipping option sync while a metaproxy provision restart is pending") - continue - } + }) } }