fix(accounts): prevent duplicate retries from crossing admins

Ambiguous idempotency-store failures can occur after the account transaction commits. Scope durable recovery markers to the authenticated admin, retain the operation key across reloads, and recover only an already committed copy without rerunning active work.

Constraint: Generic idempotent handlers may legitimately remain active while a recovery lookup is attempted

Rejected: Reclaim or rerun an in-progress duplicate request | can execute account creation concurrently

Rejected: Recover by source account and key alone | allows another admin to observe the committed copy

Confidence: high

Scope-risk: narrow

Reversibility: clean

Directive: Keep ambiguous-response recovery read-only and bind durable operation markers to the authenticated actor

Tested: Full Go unit suite, go vet, server build, integration-test compilation; frontend lint, typecheck, 1,030 Vitest tests, and production build

Not-tested: Docker-backed PostgreSQL integration runtime because Docker is unavailable

Related: Wei-Shaw/sub2api#1379

Related: Wei-Shaw/sub2api#2928
This commit is contained in:
yan9651688
2026-07-15 10:51:13 +08:00
parent 60ff61132d
commit f7da6e2bc6
9 changed files with 280 additions and 26 deletions
@@ -868,6 +868,7 @@ func (h *AccountHandler) Duplicate(c *gin.Context) {
response.BadRequest(c, "Invalid account ID")
return
}
actorScope := adminActorScope(c)
result, err := executeAdminIdempotent(
c,
@@ -877,7 +878,7 @@ func (h *AccountHandler) Duplicate(c *gin.Context) {
}{AccountID: accountID},
service.DefaultWriteIdempotencyTTL(),
func(ctx context.Context) (any, error) {
account, execErr := h.adminService.DuplicateAccount(ctx, accountID, c.GetHeader("Idempotency-Key"))
account, execErr := h.adminService.DuplicateAccount(ctx, accountID, actorScope, c.GetHeader("Idempotency-Key"))
if execErr != nil {
return nil, execErr
}
@@ -885,6 +886,17 @@ func (h *AccountHandler) Duplicate(c *gin.Context) {
},
)
if err != nil {
reason := infraerrors.Reason(err)
if reason == infraerrors.Reason(service.ErrIdempotencyInProgress) || reason == infraerrors.Reason(service.ErrIdempotencyStoreUnavail) {
recovered, recoverErr := h.adminService.RecoverDuplicateAccount(c.Request.Context(), accountID, actorScope, c.GetHeader("Idempotency-Key"))
if recoverErr != nil {
slog.Warn("account_duplicate_recovery_failed", "account_id", accountID, "actor_scope", actorScope, "reason", reason, "error", recoverErr)
} else if recovered != nil {
c.Header("X-Idempotency-Recovered", "true")
response.Success(c, h.buildAccountResponseWithRuntime(c.Request.Context(), recovered))
return
}
}
response.ErrorFrom(c, err)
return
}
@@ -5,10 +5,14 @@ package admin
import (
"context"
"encoding/json"
"errors"
"net/http"
"net/http/httptest"
"sync/atomic"
"testing"
"time"
middleware2 "github.com/Wei-Shaw/sub2api/internal/server/middleware"
"github.com/Wei-Shaw/sub2api/internal/service"
"github.com/gin-gonic/gin"
"github.com/stretchr/testify/require"
@@ -18,18 +22,74 @@ type duplicateAccountAdminServiceStub struct {
service.AdminService
account *service.Account
calls int
recoverCalls int
accountID int64
actorScope string
operationKey string
recoverScope string
recoverKey string
recoverErr error
created bool
}
func (s *duplicateAccountAdminServiceStub) DuplicateAccount(_ context.Context, accountID int64, operationKey string) (*service.Account, error) {
type blockingDuplicateAdminServiceStub struct {
service.AdminService
account *service.Account
started chan struct{}
release chan struct{}
calls atomic.Int32
recoverCalls atomic.Int32
recoverErr error
}
type failOnceMarkSucceededRepo struct {
*memoryIdempotencyRepoStub
failNext bool
}
func (r *failOnceMarkSucceededRepo) MarkSucceeded(ctx context.Context, id int64, responseStatus int, responseBody string, expiresAt time.Time) error {
if r.failNext {
r.failNext = false
return errors.New("mark succeeded failed")
}
return r.memoryIdempotencyRepoStub.MarkSucceeded(ctx, id, responseStatus, responseBody, expiresAt)
}
func (s *duplicateAccountAdminServiceStub) DuplicateAccount(_ context.Context, accountID int64, actorScope, operationKey string) (*service.Account, error) {
s.calls++
s.accountID = accountID
s.actorScope = actorScope
s.operationKey = operationKey
s.created = true
return s.account, nil
}
func setupDuplicateAccountRouter(t *testing.T, svc *duplicateAccountAdminServiceStub) *gin.Engine {
func (s *duplicateAccountAdminServiceStub) RecoverDuplicateAccount(_ context.Context, _ int64, actorScope, operationKey string) (*service.Account, error) {
s.recoverCalls++
s.recoverScope = actorScope
s.recoverKey = operationKey
if s.recoverErr != nil {
return nil, s.recoverErr
}
if !s.created {
return nil, nil
}
return s.account, nil
}
func (s *blockingDuplicateAdminServiceStub) DuplicateAccount(_ context.Context, _ int64, _, _ string) (*service.Account, error) {
s.calls.Add(1)
close(s.started)
<-s.release
return s.account, nil
}
func (s *blockingDuplicateAdminServiceStub) RecoverDuplicateAccount(_ context.Context, _ int64, _, _ string) (*service.Account, error) {
s.recoverCalls.Add(1)
return nil, s.recoverErr
}
func setupDuplicateAccountRouter(t *testing.T, svc service.AdminService) *gin.Engine {
t.Helper()
previousCoordinator := service.DefaultIdempotencyCoordinator()
service.SetDefaultIdempotencyCoordinator(nil)
@@ -37,6 +97,10 @@ func setupDuplicateAccountRouter(t *testing.T, svc *duplicateAccountAdminService
gin.SetMode(gin.TestMode)
router := gin.New()
router.Use(func(c *gin.Context) {
c.Set(string(middleware2.ContextKeyUser), middleware2.AuthSubject{UserID: 77})
c.Next()
})
handler := NewAccountHandler(svc, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil)
router.POST("/api/v1/admin/accounts/:id/duplicate", handler.Duplicate)
return router
@@ -117,6 +181,115 @@ func TestDuplicateAccountHandlerReplaysSameIdempotencyKey(t *testing.T) {
require.Equal(t, http.StatusOK, second.Code)
require.Equal(t, 1, svc.calls)
require.Equal(t, int64(42), svc.accountID)
require.Equal(t, "admin:77", svc.actorScope)
require.Equal(t, "duplicate-account-42", svc.operationKey)
require.Equal(t, "true", second.Header().Get("X-Idempotency-Replayed"))
}
func TestDuplicateAccountHandlerRecoversAfterMarkSucceededFailure(t *testing.T) {
svc := &duplicateAccountAdminServiceStub{
account: &service.Account{
ID: 43,
Name: "primary (Copy)",
Platform: service.PlatformAnthropic,
Type: service.AccountTypeAPIKey,
Status: service.StatusActive,
Schedulable: false,
},
}
router := setupDuplicateAccountRouter(t, svc)
repo := &failOnceMarkSucceededRepo{memoryIdempotencyRepoStub: newMemoryIdempotencyRepoStub(), failNext: true}
service.SetDefaultIdempotencyCoordinator(service.NewIdempotencyCoordinator(repo, service.DefaultIdempotencyConfig()))
call := func() *httptest.ResponseRecorder {
recorder := httptest.NewRecorder()
request := httptest.NewRequest(http.MethodPost, "/api/v1/admin/accounts/42/duplicate", nil)
request.Header.Set("Idempotency-Key", "duplicate-account-42-recovery")
router.ServeHTTP(recorder, request)
return recorder
}
first := call()
second := call()
require.Equal(t, http.StatusOK, first.Code)
require.Equal(t, http.StatusOK, second.Code)
require.Equal(t, "true", first.Header().Get("X-Idempotency-Recovered"))
require.Equal(t, "true", second.Header().Get("X-Idempotency-Recovered"))
require.Equal(t, 1, svc.calls, "ambiguous retries must not repeat the create side effect")
require.Equal(t, 2, svc.recoverCalls)
require.Equal(t, "admin:77", svc.recoverScope)
require.Equal(t, "duplicate-account-42-recovery", svc.recoverKey)
require.Contains(t, second.Body.String(), `"id":43`)
}
func TestDuplicateAccountHandlerPreservesIdempotencyErrorWhenRecoveryLookupFails(t *testing.T) {
svc := &duplicateAccountAdminServiceStub{
account: &service.Account{
ID: 43,
Name: "primary (Copy)",
Platform: service.PlatformAnthropic,
Type: service.AccountTypeAPIKey,
Status: service.StatusActive,
Schedulable: false,
},
recoverErr: errors.New("recovery database unavailable"),
}
router := setupDuplicateAccountRouter(t, svc)
repo := &failOnceMarkSucceededRepo{memoryIdempotencyRepoStub: newMemoryIdempotencyRepoStub(), failNext: true}
service.SetDefaultIdempotencyCoordinator(service.NewIdempotencyCoordinator(repo, service.DefaultIdempotencyConfig()))
recorder := httptest.NewRecorder()
request := httptest.NewRequest(http.MethodPost, "/api/v1/admin/accounts/42/duplicate", nil)
request.Header.Set("Idempotency-Key", "duplicate-account-42-recovery-error")
router.ServeHTTP(recorder, request)
require.Equal(t, http.StatusServiceUnavailable, recorder.Code)
require.Contains(t, recorder.Body.String(), "IDEMPOTENCY_STORE_UNAVAILABLE")
require.Equal(t, 1, svc.calls)
require.Equal(t, 1, svc.recoverCalls)
require.Equal(t, "admin:77", svc.recoverScope)
}
func TestDuplicateAccountHandlerDoesNotReexecuteWhileOriginalIsProcessing(t *testing.T) {
svc := &blockingDuplicateAdminServiceStub{
account: &service.Account{
ID: 43,
Name: "primary (Copy)",
Platform: service.PlatformAnthropic,
Type: service.AccountTypeAPIKey,
Status: service.StatusActive,
Schedulable: false,
},
started: make(chan struct{}),
release: make(chan struct{}),
recoverErr: errors.New("recovery database unavailable"),
}
router := setupDuplicateAccountRouter(t, svc)
service.SetDefaultIdempotencyCoordinator(service.NewIdempotencyCoordinator(newMemoryIdempotencyRepoStub(), service.DefaultIdempotencyConfig()))
call := func() *httptest.ResponseRecorder {
recorder := httptest.NewRecorder()
request := httptest.NewRequest(http.MethodPost, "/api/v1/admin/accounts/42/duplicate", nil)
request.Header.Set("Idempotency-Key", "duplicate-account-42-active")
router.ServeHTTP(recorder, request)
return recorder
}
firstDone := make(chan *httptest.ResponseRecorder, 1)
go func() { firstDone <- call() }()
<-svc.started
second := call()
require.Equal(t, http.StatusConflict, second.Code)
require.Contains(t, second.Body.String(), "IDEMPOTENCY_IN_PROGRESS")
require.Equal(t, int32(1), svc.calls.Load())
require.Equal(t, int32(1), svc.recoverCalls.Load())
close(svc.release)
select {
case first := <-firstDone:
require.Equal(t, http.StatusOK, first.Code)
case <-time.After(time.Second):
t.Fatal("original duplicate request did not finish")
}
}
@@ -418,11 +418,15 @@ func (s *stubAdminService) CreateAccount(ctx context.Context, input *service.Cre
return &account, nil
}
func (s *stubAdminService) DuplicateAccount(ctx context.Context, id int64, operationKey string) (*service.Account, error) {
func (s *stubAdminService) DuplicateAccount(ctx context.Context, id int64, actorScope, operationKey string) (*service.Account, error) {
account := service.Account{ID: 301, Name: "account (Copy)", Status: service.StatusActive, Schedulable: false}
return &account, nil
}
func (s *stubAdminService) RecoverDuplicateAccount(ctx context.Context, id int64, actorScope, operationKey string) (*service.Account, error) {
return nil, nil
}
func (s *stubAdminService) UpdateAccount(ctx context.Context, id int64, input *service.UpdateAccountInput) (*service.Account, error) {
s.updateAccountCalls++
if s.updateAccountErr != nil {
@@ -37,14 +37,9 @@ func executeAdminIdempotent(
return &service.IdempotencyExecuteResult{Data: data}, nil
}
actorScope := "admin:0"
if subject, ok := middleware2.GetAuthSubjectFromContext(c); ok {
actorScope = "admin:" + strconv.FormatInt(subject.UserID, 10)
}
return coordinator.Execute(c.Request.Context(), service.IdempotencyExecuteOptions{
Scope: scope,
ActorScope: actorScope,
ActorScope: adminActorScope(c),
Method: c.Request.Method,
Route: c.FullPath(),
IdempotencyKey: c.GetHeader("Idempotency-Key"),
@@ -54,6 +49,14 @@ func executeAdminIdempotent(
}, execute)
}
func adminActorScope(c *gin.Context) string {
actorScope := "admin:0"
if subject, ok := middleware2.GetAuthSubjectFromContext(c); ok {
actorScope = "admin:" + strconv.FormatInt(subject.UserID, 10)
}
return actorScope
}
func executeAdminIdempotentJSON(
c *gin.Context,
scope string,
+16 -5
View File
@@ -184,12 +184,16 @@ func duplicateAccountGroups(source *Account) ([]AccountGroup, []int64) {
return groups, groupIDs
}
func duplicateAccountOperationID(sourceID int64, operationKey string) string {
func duplicateAccountOperationID(sourceID int64, actorScope, operationKey string) string {
operationKey = strings.TrimSpace(operationKey)
if operationKey == "" {
return ""
}
payload := "admin.accounts.duplicate\x00" + strconv.FormatInt(sourceID, 10) + "\x00" + operationKey
actorScope = strings.TrimSpace(actorScope)
if actorScope == "" {
actorScope = "admin:0"
}
payload := "admin.accounts.duplicate\x00" + actorScope + "\x00" + strconv.FormatInt(sourceID, 10) + "\x00" + operationKey
digest := sha256.Sum256([]byte(payload))
return fmt.Sprintf("%x", digest)
}
@@ -209,6 +213,13 @@ func (s *adminServiceImpl) findDuplicateByOperationID(ctx context.Context, opera
return &account, nil
}
// RecoverDuplicateAccount performs a read-only lookup for an already committed duplicate.
// It is used when the idempotency coordinator cannot confirm whether response persistence
// succeeded, and deliberately never repeats the create side effect.
func (s *adminServiceImpl) RecoverDuplicateAccount(ctx context.Context, id int64, actorScope, operationKey string) (*Account, error) {
return s.findDuplicateByOperationID(ctx, duplicateAccountOperationID(id, actorScope, operationKey))
}
func cloneAccountValuePointer[T any](value *T) *T {
if value == nil {
return nil
@@ -221,9 +232,9 @@ func cloneAccountValuePointer[T any](value *T) *T {
// runtime state. Credentials and extra configuration are deep-copied so normalization of the new
// account cannot mutate the in-memory source. Linked credential shadows are excluded because they
// intentionally do not own credentials and must be created through CreateShadow.
func (s *adminServiceImpl) DuplicateAccount(ctx context.Context, id int64, operationKey string) (*Account, error) {
operationID := duplicateAccountOperationID(id, operationKey)
existing, err := s.findDuplicateByOperationID(ctx, operationID)
func (s *adminServiceImpl) DuplicateAccount(ctx context.Context, id int64, actorScope, operationKey string) (*Account, error) {
operationID := duplicateAccountOperationID(id, actorScope, operationKey)
existing, err := s.RecoverDuplicateAccount(ctx, id, actorScope, operationKey)
if err != nil {
return nil, err
}
+4 -1
View File
@@ -69,7 +69,10 @@ type AdminService interface {
CreateAccount(ctx context.Context, input *CreateAccountInput) (*Account, error)
// DuplicateAccount creates an independent account from an existing account's configuration.
// First-class runtime columns are intentionally reset by the normal account creation path.
DuplicateAccount(ctx context.Context, id int64, operationKey string) (*Account, error)
DuplicateAccount(ctx context.Context, id int64, actorScope, operationKey string) (*Account, error)
// RecoverDuplicateAccount returns a previously committed duplicate for an ambiguous retry.
// It never creates an account.
RecoverDuplicateAccount(ctx context.Context, id int64, actorScope, operationKey string) (*Account, error)
UpdateAccount(ctx context.Context, id int64, input *UpdateAccountInput) (*Account, error)
// UpdateAccountExtra 仅对 Extra 做 JSONB 增量合并(key 级覆盖),不会影响其它字段或运行态键。
// 用于刷新流程持久化 account_uuid / org_uuid 等少量键,避免被全量快照覆盖。
@@ -142,7 +142,7 @@ func TestDuplicateAccountCopiesConfigurationAndResetsRuntimeState(t *testing.T)
}
require.NoError(t, repo.Create(ctx, source))
duplicate, err := svc.DuplicateAccount(ctx, source.ID, "")
duplicate, err := svc.DuplicateAccount(ctx, source.ID, "admin:1", "")
require.NoError(t, err)
require.NotEqual(t, source.ID, duplicate.ID)
@@ -210,7 +210,7 @@ func TestDuplicateAccountRejectsCredentialShadow(t *testing.T) {
}
require.NoError(t, repo.Create(ctx, shadow))
_, err := svc.DuplicateAccount(ctx, shadow.ID, "")
_, err := svc.DuplicateAccount(ctx, shadow.ID, "admin:1", "")
require.Error(t, err)
require.Equal(t, http.StatusBadRequest, infraerrors.Code(err))
@@ -232,7 +232,7 @@ func TestDuplicateAccountRejectsRotatingOrUnknownCredentialTypes(t *testing.T) {
}
require.NoError(t, repo.Create(ctx, source))
_, err := svc.DuplicateAccount(ctx, source.ID, "")
_, err := svc.DuplicateAccount(ctx, source.ID, "admin:1", "")
require.Error(t, err)
require.Equal(t, http.StatusBadRequest, infraerrors.Code(err))
@@ -255,7 +255,7 @@ func TestDuplicateAccountPreservesUngroupedState(t *testing.T) {
}
require.NoError(t, repo.Create(ctx, source))
duplicate, err := svc.DuplicateAccount(ctx, source.ID, "")
duplicate, err := svc.DuplicateAccount(ctx, source.ID, "admin:1", "")
require.NoError(t, err)
require.Empty(t, duplicate.GroupIDs)
@@ -277,7 +277,7 @@ func TestDuplicateAccountAtomicCreateFailureLeavesNoOrphan(t *testing.T) {
require.NoError(t, repo.Create(ctx, source))
repo.atomicCreateErr = errors.New("group binding failed")
_, err := svc.DuplicateAccount(ctx, source.ID, "")
_, err := svc.DuplicateAccount(ctx, source.ID, "admin:1", "")
require.ErrorContains(t, err, "group binding failed")
require.Len(t, repo.accounts, 1)
@@ -302,12 +302,21 @@ func TestDuplicateAccountReturnsExistingCopyForSameOperationKey(t *testing.T) {
}
require.NoError(t, repo.Create(ctx, source))
first, err := svc.DuplicateAccount(ctx, source.ID, "stable-operation-key")
first, err := svc.DuplicateAccount(ctx, source.ID, "admin:7", "stable-operation-key")
require.NoError(t, err)
second, err := svc.DuplicateAccount(ctx, source.ID, "stable-operation-key")
second, err := svc.DuplicateAccount(ctx, source.ID, "admin:7", "stable-operation-key")
require.NoError(t, err)
recovered, err := svc.RecoverDuplicateAccount(ctx, source.ID, "admin:7", "stable-operation-key")
require.NoError(t, err)
otherAdminRecovery, err := svc.RecoverDuplicateAccount(ctx, source.ID, "admin:8", "stable-operation-key")
require.NoError(t, err)
otherAdminCopy, err := svc.DuplicateAccount(ctx, source.ID, "admin:8", "stable-operation-key")
require.NoError(t, err)
require.Equal(t, first.ID, second.ID)
require.Len(t, repo.accounts, 2)
require.Equal(t, first.ID, recovered.ID)
require.Nil(t, otherAdminRecovery, "durable recovery identity must remain scoped to the initiating admin")
require.NotEqual(t, first.ID, otherAdminCopy.ID)
require.Len(t, repo.accounts, 3)
require.NotEmpty(t, first.Extra[duplicateAccountOperationIDExtraKey])
}
@@ -12,6 +12,7 @@ import { duplicate } from '@/api/admin/accounts'
describe('admin account duplicate API', () => {
beforeEach(() => {
sessionStorage.clear()
post.mockReset()
post.mockResolvedValue({ data: { id: 43, name: 'primary (Copy)' } })
vi.spyOn(globalThis.crypto, 'randomUUID').mockReturnValue('11111111-1111-4111-8111-111111111111')
@@ -40,4 +41,19 @@ describe('admin account duplicate API', () => {
const secondHeaders = post.mock.calls[1][2].headers
expect(secondHeaders).toEqual(firstHeaders)
})
it('reuses the operation key after a page reload', async () => {
post.mockRejectedValueOnce(new Error('network timeout'))
await expect(duplicate(77)).rejects.toThrow('network timeout')
const firstHeaders = post.mock.calls[0][2].headers
vi.resetModules()
post.mockResolvedValueOnce({ data: { id: 78, name: 'reload (Copy)' } })
const { duplicate: duplicateAfterReload } = await import('@/api/admin/accounts')
await duplicateAfterReload(77)
expect(post).toHaveBeenCalledTimes(2)
expect(post.mock.calls[1][2].headers).toEqual(firstHeaders)
expect(sessionStorage.length).toBe(0)
})
})
+25 -2
View File
@@ -145,17 +145,40 @@ export async function create(accountData: CreateAccountRequest): Promise<Account
*/
const duplicateOperationKeys = new Map<number, string>()
function duplicateOperationStorageKey(id: number): string {
return `sub2api:admin:account-duplicate:${id}`
}
function getStoredDuplicateOperationKey(id: number): string | null {
try {
return globalThis.sessionStorage?.getItem(duplicateOperationStorageKey(id)) ?? null
} catch {
return null
}
}
function storeDuplicateOperationKey(id: number, key: string | null): void {
try {
if (key) globalThis.sessionStorage?.setItem(duplicateOperationStorageKey(id), key)
else globalThis.sessionStorage?.removeItem(duplicateOperationStorageKey(id))
} catch {
// In-memory retry protection still works when browser storage is unavailable.
}
}
export async function duplicate(id: number): Promise<Account> {
let idempotencyKey = duplicateOperationKeys.get(id)
let idempotencyKey = duplicateOperationKeys.get(id) ?? getStoredDuplicateOperationKey(id)
if (!idempotencyKey) {
const requestID = globalThis.crypto?.randomUUID?.() ?? `${Date.now()}-${Math.random().toString(36).slice(2)}`
idempotencyKey = `account-duplicate-${id}-${requestID}`
duplicateOperationKeys.set(id, idempotencyKey)
}
duplicateOperationKeys.set(id, idempotencyKey)
storeDuplicateOperationKey(id, idempotencyKey)
const { data } = await apiClient.post<Account>(`/admin/accounts/${id}/duplicate`, undefined, {
headers: { 'Idempotency-Key': idempotencyKey }
})
duplicateOperationKeys.delete(id)
storeDuplicateOperationKey(id, null)
return data
}