diff --git a/backend/ent/account.go b/backend/ent/account.go index 038aa7e59..5708e5bf1 100644 --- a/backend/ent/account.go +++ b/backend/ent/account.go @@ -41,6 +41,8 @@ type Account struct { ProxyID *int64 `json:"proxy_id,omitempty"` // Concurrency holds the value of the "concurrency" field. Concurrency int `json:"concurrency,omitempty"` + // ReservedConcurrency holds the value of the "reserved_concurrency" field. + ReservedConcurrency int `json:"reserved_concurrency,omitempty"` // Priority holds the value of the "priority" field. Priority int `json:"priority,omitempty"` // RateMultiplier holds the value of the "rate_multiplier" field. @@ -139,7 +141,7 @@ func (*Account) scanValues(columns []string) ([]any, error) { values[i] = new(sql.NullBool) case account.FieldRateMultiplier: values[i] = new(sql.NullFloat64) - case account.FieldID, account.FieldProxyID, account.FieldConcurrency, account.FieldPriority: + case account.FieldID, account.FieldProxyID, account.FieldConcurrency, account.FieldReservedConcurrency, account.FieldPriority: values[i] = new(sql.NullInt64) case account.FieldName, account.FieldNotes, account.FieldPlatform, account.FieldType, account.FieldStatus, account.FieldErrorMessage, account.FieldSessionWindowStatus: values[i] = new(sql.NullString) @@ -239,6 +241,12 @@ func (_m *Account) assignValues(columns []string, values []any) error { } else if value.Valid { _m.Concurrency = int(value.Int64) } + case account.FieldReservedConcurrency: + if value, ok := values[i].(*sql.NullInt64); !ok { + return fmt.Errorf("unexpected type %T for field reserved_concurrency", values[i]) + } else if value.Valid { + _m.ReservedConcurrency = int(value.Int64) + } case account.FieldPriority: if value, ok := values[i].(*sql.NullInt64); !ok { return fmt.Errorf("unexpected type %T for field priority", values[i]) @@ -427,6 +435,9 @@ func (_m *Account) String() string { builder.WriteString("concurrency=") builder.WriteString(fmt.Sprintf("%v", _m.Concurrency)) builder.WriteString(", ") + builder.WriteString("reserved_concurrency=") + builder.WriteString(fmt.Sprintf("%v", _m.ReservedConcurrency)) + builder.WriteString(", ") builder.WriteString("priority=") builder.WriteString(fmt.Sprintf("%v", _m.Priority)) builder.WriteString(", ") diff --git a/backend/ent/account/account.go b/backend/ent/account/account.go index 73c0e8c25..734d10e4e 100644 --- a/backend/ent/account/account.go +++ b/backend/ent/account/account.go @@ -37,6 +37,8 @@ const ( FieldProxyID = "proxy_id" // FieldConcurrency holds the string denoting the concurrency field in the database. FieldConcurrency = "concurrency" + // FieldReservedConcurrency holds the string denoting the reserved_concurrency field in the database. + FieldReservedConcurrency = "reserved_concurrency" // FieldPriority holds the string denoting the priority field in the database. FieldPriority = "priority" // FieldRateMultiplier holds the string denoting the rate_multiplier field in the database. @@ -117,6 +119,7 @@ var Columns = []string{ FieldExtra, FieldProxyID, FieldConcurrency, + FieldReservedConcurrency, FieldPriority, FieldRateMultiplier, FieldStatus, @@ -175,6 +178,8 @@ var ( DefaultExtra func() map[string]interface{} // DefaultConcurrency holds the default value on creation for the "concurrency" field. DefaultConcurrency int + // DefaultReservedConcurrency holds the default value on creation for the "reserved_concurrency" field. + DefaultReservedConcurrency int // DefaultPriority holds the default value on creation for the "priority" field. DefaultPriority int // DefaultRateMultiplier holds the default value on creation for the "rate_multiplier" field. @@ -244,6 +249,11 @@ func ByConcurrency(opts ...sql.OrderTermOption) OrderOption { return sql.OrderByField(FieldConcurrency, opts...).ToFunc() } +// ByReservedConcurrency orders the results by the reserved_concurrency field. +func ByReservedConcurrency(opts ...sql.OrderTermOption) OrderOption { + return sql.OrderByField(FieldReservedConcurrency, opts...).ToFunc() +} + // ByPriority orders the results by the priority field. func ByPriority(opts ...sql.OrderTermOption) OrderOption { return sql.OrderByField(FieldPriority, opts...).ToFunc() diff --git a/backend/ent/account/where.go b/backend/ent/account/where.go index dea1127a2..29cb4e7ba 100644 --- a/backend/ent/account/where.go +++ b/backend/ent/account/where.go @@ -100,6 +100,11 @@ func Concurrency(v int) predicate.Account { return predicate.Account(sql.FieldEQ(FieldConcurrency, v)) } +// ReservedConcurrency applies equality check predicate on the "reserved_concurrency" field. It's identical to ReservedConcurrencyEQ. +func ReservedConcurrency(v int) predicate.Account { + return predicate.Account(sql.FieldEQ(FieldReservedConcurrency, v)) +} + // Priority applies equality check predicate on the "priority" field. It's identical to PriorityEQ. func Priority(v int) predicate.Account { return predicate.Account(sql.FieldEQ(FieldPriority, v)) @@ -640,6 +645,46 @@ func ConcurrencyLTE(v int) predicate.Account { return predicate.Account(sql.FieldLTE(FieldConcurrency, v)) } +// ReservedConcurrencyEQ applies the EQ predicate on the "reserved_concurrency" field. +func ReservedConcurrencyEQ(v int) predicate.Account { + return predicate.Account(sql.FieldEQ(FieldReservedConcurrency, v)) +} + +// ReservedConcurrencyNEQ applies the NEQ predicate on the "reserved_concurrency" field. +func ReservedConcurrencyNEQ(v int) predicate.Account { + return predicate.Account(sql.FieldNEQ(FieldReservedConcurrency, v)) +} + +// ReservedConcurrencyIn applies the In predicate on the "reserved_concurrency" field. +func ReservedConcurrencyIn(vs ...int) predicate.Account { + return predicate.Account(sql.FieldIn(FieldReservedConcurrency, vs...)) +} + +// ReservedConcurrencyNotIn applies the NotIn predicate on the "reserved_concurrency" field. +func ReservedConcurrencyNotIn(vs ...int) predicate.Account { + return predicate.Account(sql.FieldNotIn(FieldReservedConcurrency, vs...)) +} + +// ReservedConcurrencyGT applies the GT predicate on the "reserved_concurrency" field. +func ReservedConcurrencyGT(v int) predicate.Account { + return predicate.Account(sql.FieldGT(FieldReservedConcurrency, v)) +} + +// ReservedConcurrencyGTE applies the GTE predicate on the "reserved_concurrency" field. +func ReservedConcurrencyGTE(v int) predicate.Account { + return predicate.Account(sql.FieldGTE(FieldReservedConcurrency, v)) +} + +// ReservedConcurrencyLT applies the LT predicate on the "reserved_concurrency" field. +func ReservedConcurrencyLT(v int) predicate.Account { + return predicate.Account(sql.FieldLT(FieldReservedConcurrency, v)) +} + +// ReservedConcurrencyLTE applies the LTE predicate on the "reserved_concurrency" field. +func ReservedConcurrencyLTE(v int) predicate.Account { + return predicate.Account(sql.FieldLTE(FieldReservedConcurrency, v)) +} + // PriorityEQ applies the EQ predicate on the "priority" field. func PriorityEQ(v int) predicate.Account { return predicate.Account(sql.FieldEQ(FieldPriority, v)) diff --git a/backend/ent/account_create.go b/backend/ent/account_create.go index 42a561cf0..9b91764a2 100644 --- a/backend/ent/account_create.go +++ b/backend/ent/account_create.go @@ -139,6 +139,20 @@ func (_c *AccountCreate) SetNillableConcurrency(v *int) *AccountCreate { return _c } +// SetReservedConcurrency sets the "reserved_concurrency" field. +func (_c *AccountCreate) SetReservedConcurrency(v int) *AccountCreate { + _c.mutation.SetReservedConcurrency(v) + return _c +} + +// SetNillableReservedConcurrency sets the "reserved_concurrency" field if the given value is not nil. +func (_c *AccountCreate) SetNillableReservedConcurrency(v *int) *AccountCreate { + if v != nil { + _c.SetReservedConcurrency(*v) + } + return _c +} + // SetPriority sets the "priority" field. func (_c *AccountCreate) SetPriority(v int) *AccountCreate { _c.mutation.SetPriority(v) @@ -439,6 +453,10 @@ func (_c *AccountCreate) defaults() error { v := account.DefaultConcurrency _c.mutation.SetConcurrency(v) } + if _, ok := _c.mutation.ReservedConcurrency(); !ok { + v := account.DefaultReservedConcurrency + _c.mutation.SetReservedConcurrency(v) + } if _, ok := _c.mutation.Priority(); !ok { v := account.DefaultPriority _c.mutation.SetPriority(v) @@ -503,6 +521,9 @@ func (_c *AccountCreate) check() error { if _, ok := _c.mutation.Concurrency(); !ok { return &ValidationError{Name: "concurrency", err: errors.New(`ent: missing required field "Account.concurrency"`)} } + if _, ok := _c.mutation.ReservedConcurrency(); !ok { + return &ValidationError{Name: "reserved_concurrency", err: errors.New(`ent: missing required field "Account.reserved_concurrency"`)} + } if _, ok := _c.mutation.Priority(); !ok { return &ValidationError{Name: "priority", err: errors.New(`ent: missing required field "Account.priority"`)} } @@ -595,6 +616,10 @@ func (_c *AccountCreate) createSpec() (*Account, *sqlgraph.CreateSpec) { _spec.SetField(account.FieldConcurrency, field.TypeInt, value) _node.Concurrency = value } + if value, ok := _c.mutation.ReservedConcurrency(); ok { + _spec.SetField(account.FieldReservedConcurrency, field.TypeInt, value) + _node.ReservedConcurrency = value + } if value, ok := _c.mutation.Priority(); ok { _spec.SetField(account.FieldPriority, field.TypeInt, value) _node.Priority = value diff --git a/backend/ent/account_update.go b/backend/ent/account_update.go index 63fab096d..2e1484d1f 100644 --- a/backend/ent/account_update.go +++ b/backend/ent/account_update.go @@ -172,6 +172,27 @@ func (_u *AccountUpdate) AddConcurrency(v int) *AccountUpdate { return _u } +// SetReservedConcurrency sets the "reserved_concurrency" field. +func (_u *AccountUpdate) SetReservedConcurrency(v int) *AccountUpdate { + _u.mutation.ResetReservedConcurrency() + _u.mutation.SetReservedConcurrency(v) + return _u +} + +// SetNillableReservedConcurrency sets the "reserved_concurrency" field if the given value is not nil. +func (_u *AccountUpdate) SetNillableReservedConcurrency(v *int) *AccountUpdate { + if v != nil { + _u.SetReservedConcurrency(*v) + } + return _u +} + +// AddReservedConcurrency adds value to the "reserved_concurrency" field. +func (_u *AccountUpdate) AddReservedConcurrency(v int) *AccountUpdate { + _u.mutation.AddReservedConcurrency(v) + return _u +} + // SetPriority sets the "priority" field. func (_u *AccountUpdate) SetPriority(v int) *AccountUpdate { _u.mutation.ResetPriority() @@ -644,6 +665,12 @@ func (_u *AccountUpdate) sqlSave(ctx context.Context) (_node int, err error) { if value, ok := _u.mutation.AddedConcurrency(); ok { _spec.AddField(account.FieldConcurrency, field.TypeInt, value) } + if value, ok := _u.mutation.ReservedConcurrency(); ok { + _spec.SetField(account.FieldReservedConcurrency, field.TypeInt, value) + } + if value, ok := _u.mutation.AddedReservedConcurrency(); ok { + _spec.AddField(account.FieldReservedConcurrency, field.TypeInt, value) + } if value, ok := _u.mutation.Priority(); ok { _spec.SetField(account.FieldPriority, field.TypeInt, value) } @@ -1011,6 +1038,27 @@ func (_u *AccountUpdateOne) AddConcurrency(v int) *AccountUpdateOne { return _u } +// SetReservedConcurrency sets the "reserved_concurrency" field. +func (_u *AccountUpdateOne) SetReservedConcurrency(v int) *AccountUpdateOne { + _u.mutation.ResetReservedConcurrency() + _u.mutation.SetReservedConcurrency(v) + return _u +} + +// SetNillableReservedConcurrency sets the "reserved_concurrency" field if the given value is not nil. +func (_u *AccountUpdateOne) SetNillableReservedConcurrency(v *int) *AccountUpdateOne { + if v != nil { + _u.SetReservedConcurrency(*v) + } + return _u +} + +// AddReservedConcurrency adds value to the "reserved_concurrency" field. +func (_u *AccountUpdateOne) AddReservedConcurrency(v int) *AccountUpdateOne { + _u.mutation.AddReservedConcurrency(v) + return _u +} + // SetPriority sets the "priority" field. func (_u *AccountUpdateOne) SetPriority(v int) *AccountUpdateOne { _u.mutation.ResetPriority() @@ -1513,6 +1561,12 @@ func (_u *AccountUpdateOne) sqlSave(ctx context.Context) (_node *Account, err er if value, ok := _u.mutation.AddedConcurrency(); ok { _spec.AddField(account.FieldConcurrency, field.TypeInt, value) } + if value, ok := _u.mutation.ReservedConcurrency(); ok { + _spec.SetField(account.FieldReservedConcurrency, field.TypeInt, value) + } + if value, ok := _u.mutation.AddedReservedConcurrency(); ok { + _spec.AddField(account.FieldReservedConcurrency, field.TypeInt, value) + } if value, ok := _u.mutation.Priority(); ok { _spec.SetField(account.FieldPriority, field.TypeInt, value) } diff --git a/backend/ent/mutation.go b/backend/ent/mutation.go index 34b3268ec..f605fbfbd 100644 --- a/backend/ent/mutation.go +++ b/backend/ent/mutation.go @@ -1440,9 +1440,11 @@ type AccountMutation struct { _type *string credentials *map[string]interface{} extra *map[string]interface{} - concurrency *int - addconcurrency *int - priority *int + concurrency *int + addconcurrency *int + reserved_concurrency *int + addreserved_concurrency *int + priority *int addpriority *int rate_multiplier *float64 addrate_multiplier *float64 @@ -2025,6 +2027,62 @@ func (m *AccountMutation) ResetConcurrency() { m.addconcurrency = nil } +// SetReservedConcurrency sets the "reserved_concurrency" field. +func (m *AccountMutation) SetReservedConcurrency(i int) { + m.reserved_concurrency = &i + m.addreserved_concurrency = nil +} + +// ReservedConcurrency returns the value of the "reserved_concurrency" field in the mutation. +func (m *AccountMutation) ReservedConcurrency() (r int, exists bool) { + v := m.reserved_concurrency + if v == nil { + return + } + return *v, true +} + +// OldReservedConcurrency returns the old "reserved_concurrency" field's value of the Account entity. +// If the Account object wasn't provided to the builder, the object is fetched from the database. +// An error is returned if the mutation operation is not UpdateOne, or the database query fails. +func (m *AccountMutation) OldReservedConcurrency(ctx context.Context) (v int, err error) { + if !m.op.Is(OpUpdateOne) { + return v, errors.New("OldReservedConcurrency is only allowed on UpdateOne operations") + } + if m.id == nil || m.oldValue == nil { + return v, errors.New("OldReservedConcurrency requires an ID field in the mutation") + } + oldValue, err := m.oldValue(ctx) + if err != nil { + return v, fmt.Errorf("querying old value for OldReservedConcurrency: %w", err) + } + return oldValue.ReservedConcurrency, nil +} + +// AddReservedConcurrency adds i to the "reserved_concurrency" field. +func (m *AccountMutation) AddReservedConcurrency(i int) { + if m.addreserved_concurrency != nil { + *m.addreserved_concurrency += i + } else { + m.addreserved_concurrency = &i + } +} + +// AddedReservedConcurrency returns the value that was added to the "reserved_concurrency" field in this mutation. +func (m *AccountMutation) AddedReservedConcurrency() (r int, exists bool) { + v := m.addreserved_concurrency + if v == nil { + return + } + return *v, true +} + +// ResetReservedConcurrency resets all changes to the "reserved_concurrency" field. +func (m *AccountMutation) ResetReservedConcurrency() { + m.reserved_concurrency = nil + m.addreserved_concurrency = nil +} + // SetPriority sets the "priority" field. func (m *AccountMutation) SetPriority(i int) { m.priority = &i @@ -2889,6 +2947,9 @@ func (m *AccountMutation) Fields() []string { if m.concurrency != nil { fields = append(fields, account.FieldConcurrency) } + if m.reserved_concurrency != nil { + fields = append(fields, account.FieldReservedConcurrency) + } if m.priority != nil { fields = append(fields, account.FieldPriority) } @@ -2961,6 +3022,8 @@ func (m *AccountMutation) Field(name string) (ent.Value, bool) { return m.ProxyID() case account.FieldConcurrency: return m.Concurrency() + case account.FieldReservedConcurrency: + return m.ReservedConcurrency() case account.FieldPriority: return m.Priority() case account.FieldRateMultiplier: @@ -3020,6 +3083,8 @@ func (m *AccountMutation) OldField(ctx context.Context, name string) (ent.Value, return m.OldProxyID(ctx) case account.FieldConcurrency: return m.OldConcurrency(ctx) + case account.FieldReservedConcurrency: + return m.OldReservedConcurrency(ctx) case account.FieldPriority: return m.OldPriority(ctx) case account.FieldRateMultiplier: @@ -3134,6 +3199,13 @@ func (m *AccountMutation) SetField(name string, value ent.Value) error { } m.SetConcurrency(v) return nil + case account.FieldReservedConcurrency: + v, ok := value.(int) + if !ok { + return fmt.Errorf("unexpected type %T for field %s", value, name) + } + m.SetReservedConcurrency(v) + return nil case account.FieldPriority: v, ok := value.(int) if !ok { @@ -3243,6 +3315,9 @@ func (m *AccountMutation) AddedFields() []string { if m.addconcurrency != nil { fields = append(fields, account.FieldConcurrency) } + if m.addreserved_concurrency != nil { + fields = append(fields, account.FieldReservedConcurrency) + } if m.addpriority != nil { fields = append(fields, account.FieldPriority) } @@ -3259,6 +3334,8 @@ func (m *AccountMutation) AddedField(name string) (ent.Value, bool) { switch name { case account.FieldConcurrency: return m.AddedConcurrency() + case account.FieldReservedConcurrency: + return m.AddedReservedConcurrency() case account.FieldPriority: return m.AddedPriority() case account.FieldRateMultiplier: @@ -3279,6 +3356,13 @@ func (m *AccountMutation) AddField(name string, value ent.Value) error { } m.AddConcurrency(v) return nil + case account.FieldReservedConcurrency: + v, ok := value.(int) + if !ok { + return fmt.Errorf("unexpected type %T for field %s", value, name) + } + m.AddReservedConcurrency(v) + return nil case account.FieldPriority: v, ok := value.(int) if !ok { @@ -3428,6 +3512,9 @@ func (m *AccountMutation) ResetField(name string) error { case account.FieldConcurrency: m.ResetConcurrency() return nil + case account.FieldReservedConcurrency: + m.ResetReservedConcurrency() + return nil case account.FieldPriority: m.ResetPriority() return nil diff --git a/backend/ent/runtime/runtime.go b/backend/ent/runtime/runtime.go index d96f9a003..07eb5025e 100644 --- a/backend/ent/runtime/runtime.go +++ b/backend/ent/runtime/runtime.go @@ -185,30 +185,34 @@ func init() { accountDescConcurrency := accountFields[7].Descriptor() // account.DefaultConcurrency holds the default value on creation for the concurrency field. account.DefaultConcurrency = accountDescConcurrency.Default.(int) + // accountDescReservedConcurrency is the schema descriptor for reserved_concurrency field. + accountDescReservedConcurrency := accountFields[8].Descriptor() + // account.DefaultReservedConcurrency holds the default value on creation for the reserved_concurrency field. + account.DefaultReservedConcurrency = accountDescReservedConcurrency.Default.(int) // accountDescPriority is the schema descriptor for priority field. - accountDescPriority := accountFields[8].Descriptor() + accountDescPriority := accountFields[9].Descriptor() // account.DefaultPriority holds the default value on creation for the priority field. account.DefaultPriority = accountDescPriority.Default.(int) // accountDescRateMultiplier is the schema descriptor for rate_multiplier field. - accountDescRateMultiplier := accountFields[9].Descriptor() + accountDescRateMultiplier := accountFields[10].Descriptor() // account.DefaultRateMultiplier holds the default value on creation for the rate_multiplier field. account.DefaultRateMultiplier = accountDescRateMultiplier.Default.(float64) // accountDescStatus is the schema descriptor for status field. - accountDescStatus := accountFields[10].Descriptor() + accountDescStatus := accountFields[11].Descriptor() // account.DefaultStatus holds the default value on creation for the status field. account.DefaultStatus = accountDescStatus.Default.(string) // account.StatusValidator is a validator for the "status" field. It is called by the builders before save. account.StatusValidator = accountDescStatus.Validators[0].(func(string) error) // accountDescAutoPauseOnExpired is the schema descriptor for auto_pause_on_expired field. - accountDescAutoPauseOnExpired := accountFields[14].Descriptor() + accountDescAutoPauseOnExpired := accountFields[15].Descriptor() // account.DefaultAutoPauseOnExpired holds the default value on creation for the auto_pause_on_expired field. account.DefaultAutoPauseOnExpired = accountDescAutoPauseOnExpired.Default.(bool) // accountDescSchedulable is the schema descriptor for schedulable field. - accountDescSchedulable := accountFields[15].Descriptor() + accountDescSchedulable := accountFields[16].Descriptor() // account.DefaultSchedulable holds the default value on creation for the schedulable field. account.DefaultSchedulable = accountDescSchedulable.Default.(bool) // accountDescSessionWindowStatus is the schema descriptor for session_window_status field. - accountDescSessionWindowStatus := accountFields[21].Descriptor() + accountDescSessionWindowStatus := accountFields[22].Descriptor() // account.SessionWindowStatusValidator is a validator for the "session_window_status" field. It is called by the builders before save. account.SessionWindowStatusValidator = accountDescSessionWindowStatus.Validators[0].(func(string) error) accountgroupFields := schema.AccountGroup{}.Fields() diff --git a/backend/ent/schema/account.go b/backend/ent/schema/account.go index 1cfecc2d5..2ea4f027b 100644 --- a/backend/ent/schema/account.go +++ b/backend/ent/schema/account.go @@ -97,6 +97,12 @@ func (Account) Fields() []ent.Field { field.Int("concurrency"). Default(3), + // reserved_concurrency: 为已绑定 sticky session 预留的并发槽位数 + // 新 session 只能使用 concurrency - reserved_concurrency 个槽位 + // 默认值 0 表示不预留，完全向后兼容 + field.Int("reserved_concurrency"). + Default(0), + // priority: 账户优先级，数值越小优先级越高 // 调度器会优先使用高优先级的账户 field.Int("priority"). diff --git a/backend/internal/handler/admin/account_handler.go b/backend/internal/handler/admin/account_handler.go index 0fae04ac0..2bbe70775 100644 --- a/backend/internal/handler/admin/account_handler.go +++ b/backend/internal/handler/admin/account_handler.go @@ -90,6 +90,7 @@ type CreateAccountRequest struct { Extra map[string]any `json:"extra"` ProxyID *int64 `json:"proxy_id"` Concurrency int `json:"concurrency"` + ReservedConcurrency int `json:"reserved_concurrency"` Priority int `json:"priority"` RateMultiplier *float64 `json:"rate_multiplier"` GroupIDs []int64 `json:"group_ids"` @@ -108,6 +109,7 @@ type UpdateAccountRequest struct { Extra map[string]any `json:"extra"` ProxyID *int64 `json:"proxy_id"` Concurrency *int `json:"concurrency"` + ReservedConcurrency *int `json:"reserved_concurrency"` Priority *int `json:"priority"` RateMultiplier *float64 `json:"rate_multiplier"` Status string `json:"status" binding:"omitempty,oneof=active inactive"` @@ -123,6 +125,7 @@ type BulkUpdateAccountsRequest struct { Name string `json:"name"` ProxyID *int64 `json:"proxy_id"` Concurrency *int `json:"concurrency"` + ReservedConcurrency *int `json:"reserved_concurrency"` Priority *int `json:"priority"` RateMultiplier *float64 `json:"rate_multiplier"` Status string `json:"status" binding:"omitempty,oneof=active inactive error"` @@ -308,6 +311,7 @@ func (h *AccountHandler) Create(c *gin.Context) { Extra: req.Extra, ProxyID: req.ProxyID, Concurrency: req.Concurrency, + ReservedConcurrency: req.ReservedConcurrency, Priority: req.Priority, RateMultiplier: req.RateMultiplier, GroupIDs: req.GroupIDs, @@ -371,6 +375,7 @@ func (h *AccountHandler) Update(c *gin.Context) { Extra: req.Extra, ProxyID: req.ProxyID, Concurrency: req.Concurrency, // 指针类型，nil 表示未提供 + ReservedConcurrency: req.ReservedConcurrency, // 指针类型，nil 表示未提供 Priority: req.Priority, // 指针类型，nil 表示未提供 RateMultiplier: req.RateMultiplier, Status: req.Status, @@ -759,6 +764,7 @@ func (h *AccountHandler) BatchCreate(c *gin.Context) { Extra: item.Extra, ProxyID: item.ProxyID, Concurrency: item.Concurrency, + ReservedConcurrency: item.ReservedConcurrency, Priority: item.Priority, RateMultiplier: item.RateMultiplier, GroupIDs: item.GroupIDs, @@ -897,6 +903,7 @@ func (h *AccountHandler) BulkUpdate(c *gin.Context) { hasUpdates := req.Name != "" || req.ProxyID != nil || req.Concurrency != nil || + req.ReservedConcurrency != nil || req.Priority != nil || req.RateMultiplier != nil || req.Status != "" || @@ -915,6 +922,7 @@ func (h *AccountHandler) BulkUpdate(c *gin.Context) { Name: req.Name, ProxyID: req.ProxyID, Concurrency: req.Concurrency, + ReservedConcurrency: req.ReservedConcurrency, Priority: req.Priority, RateMultiplier: req.RateMultiplier, Status: req.Status, diff --git a/backend/internal/handler/dto/mappers.go b/backend/internal/handler/dto/mappers.go index eee5910e8..6ca56aff9 100644 --- a/backend/internal/handler/dto/mappers.go +++ b/backend/internal/handler/dto/mappers.go @@ -166,6 +166,7 @@ func AccountFromServiceShallow(a *service.Account) *Account { Extra: a.Extra, ProxyID: a.ProxyID, Concurrency: a.Concurrency, + ReservedConcurrency: a.ReservedConcurrency, Priority: a.Priority, RateMultiplier: a.BillingRateMultiplier(), Status: a.Status, diff --git a/backend/internal/handler/dto/types.go b/backend/internal/handler/dto/types.go index 0253caf79..180c0595b 100644 --- a/backend/internal/handler/dto/types.go +++ b/backend/internal/handler/dto/types.go @@ -108,6 +108,7 @@ type Account struct { Extra map[string]any `json:"extra"` ProxyID *int64 `json:"proxy_id"` Concurrency int `json:"concurrency"` + ReservedConcurrency int `json:"reserved_concurrency"` Priority int `json:"priority"` RateMultiplier float64 `json:"rate_multiplier"` Status string `json:"status"` diff --git a/backend/internal/repository/account_repo.go b/backend/internal/repository/account_repo.go index e3e702130..e39cef838 100644 --- a/backend/internal/repository/account_repo.go +++ b/backend/internal/repository/account_repo.go @@ -80,6 +80,7 @@ func (r *accountRepository) Create(ctx context.Context, account *service.Account SetCredentials(normalizeJSONMap(account.Credentials)). SetExtra(normalizeJSONMap(account.Extra)). SetConcurrency(account.Concurrency). + SetReservedConcurrency(account.ReservedConcurrency). SetPriority(account.Priority). SetStatus(account.Status). SetErrorMessage(account.ErrorMessage). @@ -323,6 +324,7 @@ func (r *accountRepository) Update(ctx context.Context, account *service.Account SetCredentials(normalizeJSONMap(account.Credentials)). SetExtra(normalizeJSONMap(account.Extra)). SetConcurrency(account.Concurrency). + SetReservedConcurrency(account.ReservedConcurrency). SetPriority(account.Priority). SetStatus(account.Status). SetErrorMessage(account.ErrorMessage). @@ -1127,6 +1129,11 @@ func (r *accountRepository) BulkUpdate(ctx context.Context, ids []int64, updates args = append(args, *updates.Concurrency) idx++ } + if updates.ReservedConcurrency != nil { + setClauses = append(setClauses, "reserved_concurrency = $"+itoa(idx)) + args = append(args, *updates.ReservedConcurrency) + idx++ + } if updates.Priority != nil { setClauses = append(setClauses, "priority = $"+itoa(idx)) args = append(args, *updates.Priority) @@ -1509,6 +1516,7 @@ func accountEntityToService(m *dbent.Account) *service.Account { Extra: copyJSONMap(m.Extra), ProxyID: m.ProxyID, Concurrency: m.Concurrency, + ReservedConcurrency: m.ReservedConcurrency, Priority: m.Priority, RateMultiplier: &rateMultiplier, Status: m.Status, diff --git a/backend/internal/service/account.go b/backend/internal/service/account.go index fa3ce738b..8a0a5c179 100644 --- a/backend/internal/service/account.go +++ b/backend/internal/service/account.go @@ -21,6 +21,9 @@ type Account struct { Extra map[string]any ProxyID *int64 Concurrency int + // ReservedConcurrency 为已绑定 sticky session 预留的并发槽位数。 + // 新 session 只能使用 Concurrency - ReservedConcurrency 个槽位，默认 0 表示不预留。 + ReservedConcurrency int Priority int // RateMultiplier 账号计费倍率（>=0，允许 0 表示该账号计费为 0）。 // 使用指针用于兼容旧版本调度缓存（Redis）中缺字段的情况：nil 表示按 1.0 处理。 @@ -77,6 +80,20 @@ func (a *Account) BillingRateMultiplier() float64 { return *a.RateMultiplier } +// EffectiveConcurrency 返回请求可用的有效并发上限。 +// isBound=true（已绑定 sticky session）→ 可用全部 Concurrency 槽位。 +// isBound=false（新 session）→ 只能用 Concurrency - ReservedConcurrency 个槽位。 +func (a *Account) EffectiveConcurrency(isBound bool) int { + if isBound || a.ReservedConcurrency <= 0 { + return a.Concurrency + } + effective := a.Concurrency - a.ReservedConcurrency + if effective <= 0 { + return 0 + } + return effective +} + func (a *Account) IsSchedulable() bool { if !a.IsActive() || !a.Schedulable { return false diff --git a/backend/internal/service/account_service.go b/backend/internal/service/account_service.go index f192fba43..a077d05ff 100644 --- a/backend/internal/service/account_service.go +++ b/backend/internal/service/account_service.go @@ -68,10 +68,11 @@ type AccountRepository interface { // AccountBulkUpdate describes the fields that can be updated in a bulk operation. // Nil pointers mean "do not change". type AccountBulkUpdate struct { - Name *string - ProxyID *int64 - Concurrency *int - Priority *int + Name *string + ProxyID *int64 + Concurrency *int + ReservedConcurrency *int + Priority *int RateMultiplier *float64 Status *string Schedulable *bool diff --git a/backend/internal/service/admin_service.go b/backend/internal/service/admin_service.go index 1f6e91e5c..a750133fc 100644 --- a/backend/internal/service/admin_service.go +++ b/backend/internal/service/admin_service.go @@ -158,16 +158,17 @@ type UpdateGroupInput struct { } type CreateAccountInput struct { - Name string - Notes *string - Platform string - Type string - Credentials map[string]any - Extra map[string]any - ProxyID *int64 - Concurrency int - Priority int - RateMultiplier *float64 // 账号计费倍率（>=0，允许 0） + Name string + Notes *string + Platform string + Type string + Credentials map[string]any + Extra map[string]any + ProxyID *int64 + Concurrency int + ReservedConcurrency int + Priority int + RateMultiplier *float64 // 账号计费倍率（>=0，允许 0） GroupIDs []int64 ExpiresAt *int64 AutoPauseOnExpired *bool @@ -186,6 +187,7 @@ type UpdateAccountInput struct { Extra map[string]any ProxyID *int64 Concurrency *int // 使用指针区分"未提供"和"设置为0" + ReservedConcurrency *int // 使用指针区分"未提供"和"设置为0" Priority *int // 使用指针区分"未提供"和"设置为0" RateMultiplier *float64 // 账号计费倍率（>=0，允许 0） Status string @@ -201,6 +203,7 @@ type BulkUpdateAccountsInput struct { Name string ProxyID *int64 Concurrency *int + ReservedConcurrency *int Priority *int RateMultiplier *float64 // 账号计费倍率（>=0，允许 0） Status string @@ -1072,17 +1075,18 @@ func (s *adminServiceImpl) CreateAccount(ctx context.Context, input *CreateAccou } account := &Account{ - Name: input.Name, - Notes: normalizeAccountNotes(input.Notes), - Platform: input.Platform, - Type: input.Type, - Credentials: input.Credentials, - Extra: input.Extra, - ProxyID: input.ProxyID, - Concurrency: input.Concurrency, - Priority: input.Priority, - Status: StatusActive, - Schedulable: true, + Name: input.Name, + Notes: normalizeAccountNotes(input.Notes), + Platform: input.Platform, + Type: input.Type, + Credentials: input.Credentials, + Extra: input.Extra, + ProxyID: input.ProxyID, + Concurrency: input.Concurrency, + ReservedConcurrency: input.ReservedConcurrency, + Priority: input.Priority, + Status: StatusActive, + Schedulable: true, } if input.ExpiresAt != nil && *input.ExpiresAt > 0 { expiresAt := time.Unix(*input.ExpiresAt, 0) @@ -1147,6 +1151,10 @@ func (s *adminServiceImpl) UpdateAccount(ctx context.Context, id int64, input *U if input.Concurrency != nil { account.Concurrency = *input.Concurrency } + // 只在指针非 nil 时更新 ReservedConcurrency（支持设置为 0） + if input.ReservedConcurrency != nil { + account.ReservedConcurrency = *input.ReservedConcurrency + } // 只在指针非 nil 时更新 Priority（支持设置为 0） if input.Priority != nil { account.Priority = *input.Priority @@ -1250,6 +1258,9 @@ func (s *adminServiceImpl) BulkUpdateAccounts(ctx context.Context, input *BulkUp if input.Concurrency != nil { repoUpdates.Concurrency = input.Concurrency } + if input.ReservedConcurrency != nil { + repoUpdates.ReservedConcurrency = input.ReservedConcurrency + } if input.Priority != nil { repoUpdates.Priority = input.Priority } diff --git a/backend/internal/service/gateway_service.go b/backend/internal/service/gateway_service.go index 4d1dbad07..fed829a37 100644 --- a/backend/internal/service/gateway_service.go +++ b/backend/internal/service/gateway_service.go @@ -967,7 +967,8 @@ func (s *GatewayService) SelectAccountWithLoadAwareness(ctx context.Context, gro return nil, err } - result, err := s.tryAcquireAccountSlot(ctx, account.ID, account.Concurrency) + isBound := stickyAccountID > 0 && stickyAccountID == account.ID + result, err := s.tryAcquireAccountSlot(ctx, account.ID, account.EffectiveConcurrency(isBound)) if err == nil && result.Acquired { // 获取槽位后检查会话限制（使用 sessionHash 作为会话标识符） if !s.checkAndRegisterSession(ctx, account, sessionHash) { @@ -995,7 +996,7 @@ func (s *GatewayService) SelectAccountWithLoadAwareness(ctx context.Context, gro Account: account, WaitPlan: &AccountWaitPlan{ AccountID: account.ID, - MaxConcurrency: account.Concurrency, + MaxConcurrency: account.EffectiveConcurrency(true), Timeout: cfg.StickySessionWaitTimeout, MaxWaiting: cfg.StickySessionMaxWaiting, }, @@ -1006,7 +1007,7 @@ func (s *GatewayService) SelectAccountWithLoadAwareness(ctx context.Context, gro Account: account, WaitPlan: &AccountWaitPlan{ AccountID: account.ID, - MaxConcurrency: account.Concurrency, + MaxConcurrency: account.EffectiveConcurrency(isBound), Timeout: cfg.FallbackWaitTimeout, MaxWaiting: cfg.FallbackMaxWaiting, }, @@ -1130,7 +1131,7 @@ func (s *GatewayService) SelectAccountWithLoadAwareness(ctx context.Context, gro (requestedModel == "" || s.isModelSupportedByAccountWithContext(ctx, stickyAccount, requestedModel)) && stickyAccount.IsSchedulableForModelWithContext(ctx, requestedModel) && s.isAccountSchedulableForWindowCost(ctx, stickyAccount, true) { // 粘性会话窗口费用检查 - result, err := s.tryAcquireAccountSlot(ctx, stickyAccountID, stickyAccount.Concurrency) + result, err := s.tryAcquireAccountSlot(ctx, stickyAccountID, stickyAccount.EffectiveConcurrency(true)) if err == nil && result.Acquired { // 会话数量限制检查 if !s.checkAndRegisterSession(ctx, stickyAccount, sessionHash) { @@ -1158,7 +1159,7 @@ func (s *GatewayService) SelectAccountWithLoadAwareness(ctx context.Context, gro Account: stickyAccount, WaitPlan: &AccountWaitPlan{ AccountID: stickyAccountID, - MaxConcurrency: stickyAccount.Concurrency, + MaxConcurrency: stickyAccount.EffectiveConcurrency(true), Timeout: cfg.StickySessionWaitTimeout, MaxWaiting: cfg.StickySessionMaxWaiting, }, @@ -1176,9 +1177,13 @@ func (s *GatewayService) SelectAccountWithLoadAwareness(ctx context.Context, gro // 2. 批量获取负载信息 routingLoads := make([]AccountWithConcurrency, 0, len(routingCandidates)) for _, acc := range routingCandidates { + effectiveConc := acc.EffectiveConcurrency(false) + if effectiveConc == 0 { + continue // 所有槽位已预留给粘性会话，跳过 + } routingLoads = append(routingLoads, AccountWithConcurrency{ ID: acc.ID, - MaxConcurrency: acc.Concurrency, + MaxConcurrency: effectiveConc, }) } routingLoadMap, _ := s.concurrencyService.GetAccountsLoadBatch(ctx, routingLoads) @@ -1220,7 +1225,7 @@ func (s *GatewayService) SelectAccountWithLoadAwareness(ctx context.Context, gro // 4. 尝试获取槽位 for _, item := range routingAvailable { - result, err := s.tryAcquireAccountSlot(ctx, item.account.ID, item.account.Concurrency) + result, err := s.tryAcquireAccountSlot(ctx, item.account.ID, item.account.EffectiveConcurrency(false)) if err == nil && result.Acquired { // 会话数量限制检查 if !s.checkAndRegisterSession(ctx, item.account, sessionHash) { @@ -1254,7 +1259,7 @@ func (s *GatewayService) SelectAccountWithLoadAwareness(ctx context.Context, gro Account: item.account, WaitPlan: &AccountWaitPlan{ AccountID: item.account.ID, - MaxConcurrency: item.account.Concurrency, + MaxConcurrency: item.account.EffectiveConcurrency(false), Timeout: cfg.StickySessionWaitTimeout, MaxWaiting: cfg.StickySessionMaxWaiting, }, @@ -1284,7 +1289,7 @@ func (s *GatewayService) SelectAccountWithLoadAwareness(ctx context.Context, gro (requestedModel == "" || s.isModelSupportedByAccountWithContext(ctx, account, requestedModel)) && account.IsSchedulableForModelWithContext(ctx, requestedModel) && s.isAccountSchedulableForWindowCost(ctx, account, true) { // 粘性会话窗口费用检查 - result, err := s.tryAcquireAccountSlot(ctx, accountID, account.Concurrency) + result, err := s.tryAcquireAccountSlot(ctx, accountID, account.EffectiveConcurrency(true)) if err == nil && result.Acquired { // 会话数量限制检查 // Session count limit check @@ -1311,7 +1316,7 @@ func (s *GatewayService) SelectAccountWithLoadAwareness(ctx context.Context, gro Account: account, WaitPlan: &AccountWaitPlan{ AccountID: accountID, - MaxConcurrency: account.Concurrency, + MaxConcurrency: account.EffectiveConcurrency(true), Timeout: cfg.StickySessionWaitTimeout, MaxWaiting: cfg.StickySessionMaxWaiting, }, @@ -1358,9 +1363,13 @@ func (s *GatewayService) SelectAccountWithLoadAwareness(ctx context.Context, gro accountLoads := make([]AccountWithConcurrency, 0, len(candidates)) for _, acc := range candidates { + effectiveConc := acc.EffectiveConcurrency(false) + if effectiveConc == 0 { + continue // 所有槽位已预留给粘性会话，跳过 + } accountLoads = append(accountLoads, AccountWithConcurrency{ ID: acc.ID, - MaxConcurrency: acc.Concurrency, + MaxConcurrency: effectiveConc, }) } @@ -1396,7 +1405,7 @@ func (s *GatewayService) SelectAccountWithLoadAwareness(ctx context.Context, gro break } - result, err := s.tryAcquireAccountSlot(ctx, selected.account.ID, selected.account.Concurrency) + result, err := s.tryAcquireAccountSlot(ctx, selected.account.ID, selected.account.EffectiveConcurrency(false)) if err == nil && result.Acquired { // 会话数量限制检查 if !s.checkAndRegisterSession(ctx, selected.account, sessionHash) { @@ -1428,6 +1437,10 @@ func (s *GatewayService) SelectAccountWithLoadAwareness(ctx context.Context, gro // ============ Layer 3: 兜底排队 ============ s.sortCandidatesForFallback(candidates, preferOAuth, cfg.FallbackSelectionMode) for _, acc := range candidates { + effectiveConc := acc.EffectiveConcurrency(false) + if effectiveConc == 0 { + continue // 所有槽位已预留给粘性会话，跳过 + } // 会话数量限制检查（等待计划也需要占用会话配额） if !s.checkAndRegisterSession(ctx, acc, sessionHash) { continue // 会话限制已满，尝试下一个账号 @@ -1436,7 +1449,7 @@ func (s *GatewayService) SelectAccountWithLoadAwareness(ctx context.Context, gro Account: acc, WaitPlan: &AccountWaitPlan{ AccountID: acc.ID, - MaxConcurrency: acc.Concurrency, + MaxConcurrency: effectiveConc, Timeout: cfg.FallbackWaitTimeout, MaxWaiting: cfg.FallbackMaxWaiting, }, @@ -1450,7 +1463,7 @@ func (s *GatewayService) tryAcquireByLegacyOrder(ctx context.Context, candidates sortAccountsByPriorityAndLastUsed(ordered, preferOAuth) for _, acc := range ordered { - result, err := s.tryAcquireAccountSlot(ctx, acc.ID, acc.Concurrency) + result, err := s.tryAcquireAccountSlot(ctx, acc.ID, acc.EffectiveConcurrency(false)) if err == nil && result.Acquired { // 会话数量限制检查 if !s.checkAndRegisterSession(ctx, acc, sessionHash) { @@ -1752,6 +1765,9 @@ func (s *GatewayService) tryAcquireAccountSlot(ctx context.Context, accountID in if s.concurrencyService == nil { return &AcquireResult{Acquired: true, ReleaseFunc: func() {}}, nil } + if maxConcurrency == 0 { + return &AcquireResult{Acquired: false, ReleaseFunc: nil}, nil + } return s.concurrencyService.AcquireAccountSlot(ctx, accountID, maxConcurrency) } diff --git a/backend/internal/service/openai_gateway_service.go b/backend/internal/service/openai_gateway_service.go index 6c4fe256c..4fe539a74 100644 --- a/backend/internal/service/openai_gateway_service.go +++ b/backend/internal/service/openai_gateway_service.go @@ -443,7 +443,8 @@ func (s *OpenAIGatewayService) SelectAccountWithLoadAwareness(ctx context.Contex if err != nil { return nil, err } - result, err := s.tryAcquireAccountSlot(ctx, account.ID, account.Concurrency) + isBound := stickyAccountID > 0 && stickyAccountID == account.ID + result, err := s.tryAcquireAccountSlot(ctx, account.ID, account.EffectiveConcurrency(isBound)) if err == nil && result.Acquired { return &AccountSelectionResult{ Account: account, @@ -451,14 +452,14 @@ func (s *OpenAIGatewayService) SelectAccountWithLoadAwareness(ctx context.Contex ReleaseFunc: result.ReleaseFunc, }, nil } - if stickyAccountID > 0 && stickyAccountID == account.ID && s.concurrencyService != nil { + if isBound && s.concurrencyService != nil { waitingCount, _ := s.concurrencyService.GetAccountWaitingCount(ctx, account.ID) if waitingCount < cfg.StickySessionMaxWaiting { return &AccountSelectionResult{ Account: account, WaitPlan: &AccountWaitPlan{ AccountID: account.ID, - MaxConcurrency: account.Concurrency, + MaxConcurrency: account.EffectiveConcurrency(true), Timeout: cfg.StickySessionWaitTimeout, MaxWaiting: cfg.StickySessionMaxWaiting, }, @@ -469,7 +470,7 @@ func (s *OpenAIGatewayService) SelectAccountWithLoadAwareness(ctx context.Contex Account: account, WaitPlan: &AccountWaitPlan{ AccountID: account.ID, - MaxConcurrency: account.Concurrency, + MaxConcurrency: account.EffectiveConcurrency(isBound), Timeout: cfg.FallbackWaitTimeout, MaxWaiting: cfg.FallbackMaxWaiting, }, @@ -504,7 +505,7 @@ func (s *OpenAIGatewayService) SelectAccountWithLoadAwareness(ctx context.Contex } if !clearSticky && account.IsSchedulable() && account.IsOpenAI() && (requestedModel == "" || account.IsModelSupported(requestedModel)) { - result, err := s.tryAcquireAccountSlot(ctx, accountID, account.Concurrency) + result, err := s.tryAcquireAccountSlot(ctx, accountID, account.EffectiveConcurrency(true)) if err == nil && result.Acquired { _ = s.cache.RefreshSessionTTL(ctx, derefGroupID(groupID), "openai:"+sessionHash, openaiStickySessionTTL) return &AccountSelectionResult{ @@ -520,7 +521,7 @@ func (s *OpenAIGatewayService) SelectAccountWithLoadAwareness(ctx context.Contex Account: account, WaitPlan: &AccountWaitPlan{ AccountID: accountID, - MaxConcurrency: account.Concurrency, + MaxConcurrency: account.EffectiveConcurrency(true), Timeout: cfg.StickySessionWaitTimeout, MaxWaiting: cfg.StickySessionMaxWaiting, }, @@ -556,9 +557,13 @@ func (s *OpenAIGatewayService) SelectAccountWithLoadAwareness(ctx context.Contex accountLoads := make([]AccountWithConcurrency, 0, len(candidates)) for _, acc := range candidates { + effectiveConc := acc.EffectiveConcurrency(false) + if effectiveConc == 0 { + continue // 所有槽位已预留给粘性会话，跳过 + } accountLoads = append(accountLoads, AccountWithConcurrency{ ID: acc.ID, - MaxConcurrency: acc.Concurrency, + MaxConcurrency: effectiveConc, }) } @@ -567,7 +572,7 @@ func (s *OpenAIGatewayService) SelectAccountWithLoadAwareness(ctx context.Contex ordered := append([]*Account(nil), candidates...) sortAccountsByPriorityAndLastUsed(ordered, false) for _, acc := range ordered { - result, err := s.tryAcquireAccountSlot(ctx, acc.ID, acc.Concurrency) + result, err := s.tryAcquireAccountSlot(ctx, acc.ID, acc.EffectiveConcurrency(false)) if err == nil && result.Acquired { if sessionHash != "" { _ = s.cache.SetSessionAccountID(ctx, derefGroupID(groupID), "openai:"+sessionHash, acc.ID, openaiStickySessionTTL) @@ -617,7 +622,7 @@ func (s *OpenAIGatewayService) SelectAccountWithLoadAwareness(ctx context.Contex shuffleWithinSortGroups(available) for _, item := range available { - result, err := s.tryAcquireAccountSlot(ctx, item.account.ID, item.account.Concurrency) + result, err := s.tryAcquireAccountSlot(ctx, item.account.ID, item.account.EffectiveConcurrency(false)) if err == nil && result.Acquired { if sessionHash != "" { _ = s.cache.SetSessionAccountID(ctx, derefGroupID(groupID), "openai:"+sessionHash, item.account.ID, openaiStickySessionTTL) @@ -635,11 +640,15 @@ func (s *OpenAIGatewayService) SelectAccountWithLoadAwareness(ctx context.Contex // ============ Layer 3: Fallback wait ============ sortAccountsByPriorityAndLastUsed(candidates, false) for _, acc := range candidates { + effectiveConc := acc.EffectiveConcurrency(false) + if effectiveConc == 0 { + continue // 所有槽位已预留给粘性会话，跳过 + } return &AccountSelectionResult{ Account: acc, WaitPlan: &AccountWaitPlan{ AccountID: acc.ID, - MaxConcurrency: acc.Concurrency, + MaxConcurrency: effectiveConc, Timeout: cfg.FallbackWaitTimeout, MaxWaiting: cfg.FallbackMaxWaiting, }, @@ -673,6 +682,9 @@ func (s *OpenAIGatewayService) tryAcquireAccountSlot(ctx context.Context, accoun if s.concurrencyService == nil { return &AcquireResult{Acquired: true, ReleaseFunc: func() {}}, nil } + if maxConcurrency == 0 { + return &AcquireResult{Acquired: false, ReleaseFunc: nil}, nil + } return s.concurrencyService.AcquireAccountSlot(ctx, accountID, maxConcurrency) } diff --git a/backend/migrations/058_add_account_reserved_concurrency.sql b/backend/migrations/058_add_account_reserved_concurrency.sql new file mode 100644 index 000000000..5278abb32 --- /dev/null +++ b/backend/migrations/058_add_account_reserved_concurrency.sql @@ -0,0 +1 @@ +ALTER TABLE accounts ADD COLUMN IF NOT EXISTS reserved_concurrency INTEGER NOT NULL DEFAULT 0; diff --git a/frontend/src/components/account/BulkEditAccountModal.vue b/frontend/src/components/account/BulkEditAccountModal.vue index 67de56973..c3719317a 100644 --- a/frontend/src/components/account/BulkEditAccountModal.vue +++ b/frontend/src/components/account/BulkEditAccountModal.vue @@ -459,7 +459,7 @@ -
+
+
+
+ + +
+ +

{{ t('admin.accounts.reservedConcurrencyHint') }}

+
+
+ + +

{{ t('admin.accounts.reservedConcurrencyHint') }}

+
, proxy_id: null as number | null, concurrency: 10, + reserved_concurrency: 0, priority: 1, rate_multiplier: 1, group_ids: [] as number[], @@ -2608,6 +2614,7 @@ const resetForm = () => { form.credentials = {} form.proxy_id = null form.concurrency = 10 + form.reserved_concurrency = 0 form.priority = 1 form.rate_multiplier = 1 form.group_ids = [] @@ -2876,6 +2883,7 @@ const createAccountAndFinish = async ( extra, proxy_id: form.proxy_id, concurrency: form.concurrency, + reserved_concurrency: form.reserved_concurrency, priority: form.priority, rate_multiplier: form.rate_multiplier, group_ids: form.group_ids, @@ -2964,6 +2972,7 @@ const handleOpenAIValidateRT = async (refreshTokenInput: string) => { extra, proxy_id: form.proxy_id, concurrency: form.concurrency, + reserved_concurrency: form.reserved_concurrency, priority: form.priority, rate_multiplier: form.rate_multiplier, group_ids: form.group_ids, @@ -3053,6 +3062,7 @@ const handleAntigravityValidateRT = async (refreshTokenInput: string) => { extra: {}, proxy_id: form.proxy_id, concurrency: form.concurrency, + reserved_concurrency: form.reserved_concurrency, priority: form.priority, rate_multiplier: form.rate_multiplier, group_ids: form.group_ids, @@ -3344,6 +3354,7 @@ const handleCookieAuth = async (sessionKey: string) => { extra, proxy_id: form.proxy_id, concurrency: form.concurrency, + reserved_concurrency: form.reserved_concurrency, priority: form.priority, rate_multiplier: form.rate_multiplier, group_ids: form.group_ids, diff --git a/frontend/src/components/account/EditAccountModal.vue b/frontend/src/components/account/EditAccountModal.vue index ed243276f..b428a8b73 100644 --- a/frontend/src/components/account/EditAccountModal.vue +++ b/frontend/src/components/account/EditAccountModal.vue @@ -672,6 +672,11 @@
+
+ + +

{{ t('admin.accounts.reservedConcurrencyHint') }}

+
) proxy_id: number | null concurrency: number + reserved_concurrency: number current_concurrency?: number // Real-time concurrency count from Redis priority: number rate_multiplier?: number // Account billing multiplier (>=0, 0 means free) @@ -693,6 +694,7 @@ export interface CreateAccountRequest { extra?: Record proxy_id?: number | null concurrency?: number + reserved_concurrency?: number priority?: number rate_multiplier?: number // Account billing multiplier (>=0, 0 means free) group_ids?: number[] @@ -709,6 +711,7 @@ export interface UpdateAccountRequest { extra?: Record proxy_id?: number | null concurrency?: number + reserved_concurrency?: number priority?: number rate_multiplier?: number // Account billing multiplier (>=0, 0 means free) schedulable?: boolean