fix(billing): durable usage recovery and canonical monetary accounting

backend/internal/handler/gateway_billing_delivery_http_test.go

@@ -0,0 +1,82 @@
+//go:build unit
+
+package handler
+
+import (
+	"context"
+	"errors"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"sync/atomic"
+	"testing"
+
+	"github.com/Wei-Shaw/sub2api/internal/service"
+	"github.com/gin-gonic/gin"
+	"github.com/stretchr/testify/require"
+)
+
+type httpDeliveryBilling struct {
+	service.UsageBillingRepository
+	event        *service.UsageBillingEvent
+	failure      error
+	stageFailure error
+	applied      bool
+}
+
+func (r *httpDeliveryBilling) StageUsageBilling(_ context.Context, c *service.UsageBillingCommand, u *service.UsageLog) error {
+	if r.stageFailure != nil {
+		return r.stageFailure
+	}
+	r.event = &service.UsageBillingEvent{Version: 1, Command: *c, Usage: *u}
+	return nil
+}
+func (r *httpDeliveryBilling) ApplyStagedUsageBilling(context.Context, service.UsageBillingEventKey) (*service.UsageBillingRecovery, error) {
+	if r.failure != nil {
+		return nil, r.failure
+	}
+	fresh := !r.applied
+	r.applied = true
+	return &service.UsageBillingRecovery{Event: *r.event, Result: service.UsageBillingApplyResult{Applied: fresh}, NeedsCache: true}, nil
+}
+func (*httpDeliveryBilling) PendingUsageBilling(context.Context, int) ([]service.UsageBillingEventKey, error) {
+	return nil, nil
+}
+func (*httpDeliveryBilling) AcknowledgeUsageBillingCache(context.Context, service.UsageBillingEventKey) error {
+	return nil
+}
+
+// Real HTTP transport + production protocol adapter + GatewayHandler. The SQL
+// crash/atomicity half is exercised separately with PostgreSQL integration tests.
+func TestGatewayHTTPAdapterStagesUsageBeforeExecutionAndNeverRecallsModel(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+	for _, endpoint := range []string{"responses", "chat/completions"} {
+		t.Run(endpoint, func(t *testing.T) {
+			var calls atomic.Int32
+			upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
+				calls.Add(1)
+				w.Header().Set("Content-Type", "text/event-stream")
+				_, _ = io.WriteString(w, compatStreamUsagePartialStream+compatStreamUsageCompletion)
+			}))
+			defer upstream.Close()
+			billing := &httpDeliveryBilling{failure: errors.New("injected execution failure after durable stage")}
+			run := runCompatStreamUsageHandlerWithDelivery(t, endpoint, "", true, billing, upstream.URL)
+			require.Equal(t, int32(1), calls.Load())
+			require.Contains(t, run.rec.Body.String(), "partial")
+			require.NotNil(t, billing.event, "handler must stage before returning, not queue an in-memory closure")
+			require.Equal(t, 12, billing.event.Usage.InputTokens)
+			require.Equal(t, 7, billing.event.Usage.OutputTokens)
+			require.Greater(t, billing.event.Usage.ActualCost, 0.0)
+			require.Empty(t, run.usageRepo.created, "failure is not a fake zero-cost success")
+			billing.failure = nil
+			key := service.UsageBillingEventKey{RequestID: billing.event.Command.RequestID, APIKeyID: billing.event.Command.APIKeyID}
+			first, err := billing.ApplyStagedUsageBilling(context.Background(), key)
+			require.NoError(t, err)
+			require.True(t, first.Result.Applied)
+			second, err := billing.ApplyStagedUsageBilling(context.Background(), key)
+			require.NoError(t, err)
+			require.False(t, second.Result.Applied)
+			require.Equal(t, int32(1), calls.Load(), "billing recovery cannot invoke a model transport")
+		})
+	}
+}

backend/internal/handler/gateway_compat_stream_usage_test.go

@@ -14,6 +14,7 @@ import (
 
 	"github.com/Wei-Shaw/sub2api/internal/config"
 	"github.com/Wei-Shaw/sub2api/internal/pkg/tlsfingerprint"
+	"github.com/Wei-Shaw/sub2api/internal/repository"
 	middleware "github.com/Wei-Shaw/sub2api/internal/server/middleware"
 	"github.com/Wei-Shaw/sub2api/internal/service"
 	"github.com/gin-gonic/gin"
@@ -68,8 +69,14 @@ type compatStreamUsageRun struct {
 // runCompatStreamUsageHandler 经 GatewayHandler 转发一次流式请求；上游写完 upstreamStream 后，
 // closeUpstream 为 false 时保持连接不再发送数据。
 func runCompatStreamUsageHandler(t *testing.T, endpoint, upstreamStream string, closeUpstream bool) compatStreamUsageRun {
+	return runCompatStreamUsageHandlerWithDelivery(t, endpoint, upstreamStream, closeUpstream, nil, "")
+}
+
+func runCompatStreamUsageHandlerWithDelivery(t *testing.T, endpoint, upstreamStream string, closeUpstream bool, durable service.UsageBillingRepository, baseURL string) compatStreamUsageRun {
 	t.Helper()
 	cfg := &config.Config{Gateway: config.GatewayConfig{StreamDataIntervalTimeout: 1}}
+	cfg.Security.URLAllowlist.AllowPrivateHosts = true
+	cfg.Security.URLAllowlist.AllowInsecureHTTP = true
 	groupID := int64(9300)
 	group := &service.Group{ID: groupID, Hydrated: true, Platform: service.PlatformAnthropic, Status: service.StatusActive, RateMultiplier: 1}
 	account := &service.Account{
@@ -99,9 +106,18 @@ func runCompatStreamUsageHandler(t *testing.T, endpoint, upstreamStream string,
 		StatusCode: http.StatusOK, Body: pr,
 		Header: http.Header{"X-Request-Id": []string{"compat-stream-usage"}},
 	}}
+	var billing service.UsageBillingRepository = run.billingRepo
+	var transport service.HTTPUpstream = run.upstream
+	if durable != nil {
+		billing = durable
+	}
+	if baseURL != "" {
+		account.Credentials["base_url"] = baseURL
+		transport = repository.NewHTTPUpstream(cfg)
+	}
 	gw := service.NewGatewayService(
-		nil, &fakeGroupRepo{group: group}, run.usageRepo, run.billingRepo, nil, nil, nil, nil, cfg,
-		snapshot, nil, service.NewBillingService(cfg, nil), nil, nil, nil, run.upstream,
+		nil, &fakeGroupRepo{group: group}, run.usageRepo, billing, nil, nil, nil, nil, cfg,
+		snapshot, nil, service.NewBillingService(cfg, nil), nil, nil, nil, transport,
 		&service.DeferredService{}, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil,
 	)
 	h := &GatewayHandler{

backend/internal/handler/gateway_handler.go

@@ -2493,7 +2493,7 @@ func (h *GatewayHandler) submitUsageRecordTask(parent context.Context, task serv
 		return
 	}
 	task, abandon := wrapUsageRecordTaskContext(parent, task)
-	if h.usageRecordWorkerPool != nil {
+	if h.usageRecordWorkerPool != nil && !h.gatewayService.DurableUsageBillingEnabled() {
 		if mode := h.usageRecordWorkerPool.Submit(task); mode != service.UsageRecordSubmitModeDroppedStopped {
 			if mode.Dropped() {
 				abandon()
@@ -2526,7 +2526,7 @@ func (h *GatewayHandler) submitMandatoryUsageRecordTask(parent context.Context,
 		return
 	}
 	task, _ = wrapUsageRecordTaskContext(parent, task)
-	if h.usageRecordWorkerPool != nil {
+	if h.usageRecordWorkerPool != nil && !h.gatewayService.DurableUsageBillingEnabled() {
 		if mode := h.usageRecordWorkerPool.Submit(task); !mode.Dropped() {
 			return
 		}

backend/internal/handler/openai_billing_delivery_http_test.go

@@ -0,0 +1,111 @@
+//go:build unit
+
+package handler
+
+import (
+	"context"
+	"errors"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/Wei-Shaw/sub2api/internal/config"
+	"github.com/Wei-Shaw/sub2api/internal/repository"
+	"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"
+)
+
+func TestOpenAIGatewayHandlerRealHTTPDeliveryFailureDoesNotInventUsageOrRecallModel(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+	for _, phase := range []string{"stage", "execute"} {
+		t.Run(phase, func(t *testing.T) {
+			var calls atomic.Int32
+			upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				calls.Add(1)
+				w.Header().Set("Content-Type", "application/json")
+				_, _ = io.WriteString(w, `{"id":"resp_billing_boundary","object":"response","status":"completed","model":"gpt-5.1","output":[],"usage":{"input_tokens":12,"output_tokens":7,"total_tokens":19}}`)
+			}))
+			defer upstream.Close()
+			cfg := &config.Config{}
+			cfg.Default.RateMultiplier = 1
+			cfg.Security.URLAllowlist.AllowPrivateHosts = true
+			cfg.Security.URLAllowlist.AllowInsecureHTTP = true
+			groupID := int64(9201)
+			group := &service.Group{ID: groupID, Platform: service.PlatformOpenAI, Status: service.StatusActive, Hydrated: true, RateMultiplier: 1}
+			account := service.Account{ID: 9202, Platform: service.PlatformOpenAI, Type: service.AccountTypeAPIKey,
+				Status: service.StatusActive, Schedulable: true, Concurrency: 1, GroupIDs: []int64{groupID},
+				Credentials: map[string]any{"api_key": "synthetic", "base_url": upstream.URL}}
+			accounts := &openAIWSUsageHandlerAccountRepoStub{account: account}
+			logs := &openAIWSUsageHandlerUsageLogRepoStub{created: make(chan *service.UsageLog, 2)}
+			billing := &httpDeliveryBilling{}
+			if phase == "stage" {
+				billing.stageFailure = errors.New("injected durable stage failure")
+			} else {
+				billing.failure = errors.New("injected execution failure")
+			}
+			billingCache := service.NewBillingCacheService(newHandlerInflightCache(100), nil, nil, nil, nil, nil, cfg, nil)
+			t.Cleanup(billingCache.Stop)
+			concurrency := service.NewConcurrencyService(&concurrencyCacheMock{
+				acquireUserSlotFn:    func(context.Context, int64, int, string) (bool, error) { return true, nil },
+				acquireAccountSlotFn: func(context.Context, int64, int, string) (bool, error) { return true, nil },
+			})
+			gateway := service.NewOpenAIGatewayService(accounts, nil, logs, billing, nil, nil, nil, nil, cfg, nil, concurrency,
+				service.NewBillingService(cfg, nil), nil, nil, repository.NewHTTPUpstream(cfg), &service.DeferredService{}, nil, nil, nil, nil, nil, nil, nil)
+			t.Cleanup(gateway.CloseOpenAIWSPool)
+			h := &OpenAIGatewayHandler{gatewayService: gateway, cfg: cfg, billingCacheService: billingCache,
+				apiKeyService: &service.APIKeyService{}, concurrencyHelper: NewConcurrencyHelper(concurrency, SSEPingFormatNone, 0)}
+			key := &service.APIKey{ID: 9203, UserID: 9204, GroupID: &groupID, Group: group,
+				User: &service.User{ID: 9204, Status: service.StatusActive, Balance: 100}}
+			router := gin.New()
+			router.Use(middleware.ClientRequestID(), func(c *gin.Context) {
+				c.Set(string(middleware.ContextKeyAPIKey), key)
+				c.Set(string(middleware.ContextKeyUser), middleware.AuthSubject{UserID: key.UserID, Concurrency: 1})
+				c.Next()
+			})
+			finished := make(chan struct{})
+			router.POST("/openai/v1/responses", func(c *gin.Context) {
+				defer close(finished)
+				h.Responses(c)
+			})
+			ingress := httptest.NewServer(router)
+			defer ingress.Close()
+			client := &http.Client{Timeout: 5 * time.Second}
+			res, err := client.Post(ingress.URL+"/openai/v1/responses", "application/json", strings.NewReader(`{"model":"gpt-5.1","input":"synthetic","stream":false}`))
+			require.NoError(t, err)
+			body, err := io.ReadAll(res.Body)
+			_ = res.Body.Close()
+			require.NoError(t, err)
+			select {
+			case <-finished:
+			case <-time.After(5 * time.Second):
+				t.Fatal("actual OpenAI handler did not finish its delivery boundary")
+			}
+			require.Equal(t, http.StatusOK, res.StatusCode)
+			require.Contains(t, string(body), "resp_billing_boundary")
+			require.Equal(t, int32(1), calls.Load())
+			require.Empty(t, logs.created, "neither a stage failure nor an execution failure is a successful zero-cost usage row")
+			if phase == "stage" {
+				require.Nil(t, billing.event, "pre-storage failure is an explicit unrecoverable observation boundary")
+				return
+			}
+			require.NotNil(t, billing.event, "the actual OpenAI handler must stage synchronously before it returns")
+			require.Equal(t, 12, billing.event.Usage.InputTokens)
+			require.Equal(t, 7, billing.event.Usage.OutputTokens)
+			require.Greater(t, billing.event.Usage.ActualCost, 0.0)
+			billing.failure = nil
+			eventKey := service.UsageBillingEventKey{RequestID: billing.event.Command.RequestID, APIKeyID: key.ID}
+			for repeat := 0; repeat < 2; repeat++ {
+				result, err := billing.ApplyStagedUsageBilling(context.Background(), eventKey)
+				require.NoError(t, err)
+				require.Equal(t, repeat == 0, result.Result.Applied)
+			}
+			require.Equal(t, int32(1), calls.Load())
+		})
+	}
+}

backend/internal/handler/openai_gateway_handler.go

@@ -4022,7 +4022,7 @@ func (h *OpenAIGatewayHandler) submitUsageRecordTask(parent context.Context, tas
 		return
 	}
 	task, abandon := wrapUsageRecordTaskContext(parent, task)
-	if h.usageRecordWorkerPool != nil {
+	if h.usageRecordWorkerPool != nil && !h.gatewayService.DurableUsageBillingEnabled() {
 		if mode := h.usageRecordWorkerPool.Submit(task); mode != service.UsageRecordSubmitModeDroppedStopped {
 			if mode.Dropped() {
 				abandon()
@@ -4064,7 +4064,7 @@ func (h *OpenAIGatewayHandler) submitMandatoryUsageRecordTask(parent context.Con
 		return
 	}
 	task, _ = wrapUsageRecordTaskContext(parent, task)
-	if h.usageRecordWorkerPool != nil {
+	if h.usageRecordWorkerPool != nil && !h.gatewayService.DurableUsageBillingEnabled() {
 		if mode := h.usageRecordWorkerPool.Submit(task); !mode.Dropped() {
 			return
 		}