feat: hot-reload provision releases instead of restarting

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,
 		},
 	})
 }

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

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

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
 }

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,14 +32,24 @@ 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()
 	})
 	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,14 +214,17 @@ 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)
 	require.Equal(t, "old-key", channel.Key)
 	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()
 }

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