Reliability: burst capacity and batched relay ingestion

internal/v4/httpapi/relay.go

@@ -9,7 +9,10 @@ import (
 	"github.com/Phloraxx/payment-api/internal/v4/relay"
 )
 
-const relayHTTPBodyLimit = 64 << 10
+const (
+	relayHTTPBodyLimit      = 64 << 10
+	relayHTTPBatchBodyLimit = 1 << 20
+)
 
 const (
 	relayDeviceHeader    = "X-PayGate-Relay-Device"
@@ -27,6 +30,7 @@ func NewRelayHandler(service *relay.Service) *RelayHandler {
 	h := &RelayHandler{Relay: service, mux: http.NewServeMux()}
 	h.mux.HandleFunc("POST /api/v4/relay/pair", h.pair)
 	h.mux.HandleFunc("POST "+relay.EventPath, h.event)
+	h.mux.HandleFunc("POST "+relay.EventBatchPath, h.eventBatch)
 	h.mux.HandleFunc("POST "+relay.HeartbeatPath, h.heartbeat)
 	h.mux.HandleFunc("GET "+relay.DevicePath, h.getThisDevice)
 	return h
@@ -90,6 +94,19 @@ func (h *RelayHandler) event(w http.ResponseWriter, r *http.Request) {
 	writeJSON(w, http.StatusOK, result)
 }
 
+func (h *RelayHandler) eventBatch(w http.ResponseWriter, r *http.Request) {
+	raw, ok := readRelayBodyLimit(w, r, relayHTTPBatchBodyLimit)
+	if !ok {
+		return
+	}
+	result, err := h.Relay.IngestBatchSigned(r.Context(), relayAuth(r, relay.EventBatchPath), raw)
+	if err != nil {
+		writeRelayError(w, err)
+		return
+	}
+	writeJSON(w, http.StatusOK, result)
+}
+
 func (h *RelayHandler) heartbeat(w http.ResponseWriter, r *http.Request) {
 	raw, ok := readRelayBody(w, r)
 	if !ok {
@@ -133,11 +150,15 @@ func relayAuth(r *http.Request, path string) relay.RequestAuth {
 }
 
 func readRelayBody(w http.ResponseWriter, r *http.Request) ([]byte, bool) {
+	return readRelayBodyLimit(w, r, relayHTTPBodyLimit)
+}
+
+func readRelayBodyLimit(w http.ResponseWriter, r *http.Request, limit int64) ([]byte, bool) {
 	if !isJSON(r.Header.Get("Content-Type")) {
 		writeError(w, http.StatusUnsupportedMediaType, "invalid_content_type", "Content-Type must be application/json")
 		return nil, false
 	}
-	r.Body = http.MaxBytesReader(w, r.Body, relayHTTPBodyLimit)
+	r.Body = http.MaxBytesReader(w, r.Body, limit)
 	raw, err := io.ReadAll(r.Body)
 	if err != nil {
 		writeError(w, http.StatusBadRequest, "invalid_request", "Relay body is too large or unreadable")

internal/v4/httpapi/relay_test.go

@@ -193,6 +193,45 @@ func TestRelaySignedEventAndSignatureFailure(t *testing.T) {
 	}
 }
 
+func TestRelaySignedBatchAcceptsAndDeduplicatesEvents(t *testing.T) {
+	f := newRelayHTTPFixture(t)
+	epoch := pairRelayHTTP(t, f)
+	eventA := map[string]any{
+		"schema_version": 1, "event_id": strings.Repeat("a", 64),
+		"package_name": "com.paytm.business", "posted_at_ms": f.now.UnixMilli(),
+		"title": "Payment Received on Paytm for Business", "text": "Received Rs. 100.00 from Alice",
+	}
+	eventB := map[string]any{
+		"schema_version": 1, "event_id": strings.Repeat("b", 64),
+		"package_name": "com.paytm.business", "posted_at_ms": f.now.UnixMilli(),
+		"title": "Payment Received on Paytm for Business", "text": "Received Rs. 100.00 from Bob",
+	}
+	body, _ := json.Marshal(map[string]any{"schema_version": 1, "events": []any{eventA, eventB}})
+
+	rr := httptest.NewRecorder()
+	f.handler.ServeHTTP(rr, signedRelayRequest(t, f, relay.EventBatchPath, body, epoch))
+	if rr.Code != http.StatusOK {
+		t.Fatalf("batch status=%d body=%s", rr.Code, rr.Body.String())
+	}
+	if !strings.Contains(rr.Body.String(), `"event_id":"`+strings.Repeat("a", 64)+`"`) ||
+		!strings.Contains(rr.Body.String(), `"event_id":"`+strings.Repeat("b", 64)+`"`) {
+		t.Fatalf("batch response missing source ids: %s", rr.Body.String())
+	}
+	var count int
+	if err := f.db.SQL.QueryRow(`SELECT COUNT(*) FROM relay_events`).Scan(&count); err != nil || count != 2 {
+		t.Fatalf("relay events=%d err=%v", count, err)
+	}
+
+	rr = httptest.NewRecorder()
+	f.handler.ServeHTTP(rr, signedRelayRequest(t, f, relay.EventBatchPath, body, epoch))
+	if rr.Code != http.StatusOK || strings.Count(rr.Body.String(), `"duplicate":true`) != 2 {
+		t.Fatalf("batch retry status=%d body=%s", rr.Code, rr.Body.String())
+	}
+	if err := f.db.SQL.QueryRow(`SELECT COUNT(*) FROM relay_events`).Scan(&count); err != nil || count != 2 {
+		t.Fatalf("relay events after retry=%d err=%v", count, err)
+	}
+}
+
 func TestRelayRejectsQueryParameters(t *testing.T) {
 	f := newRelayHTTPFixture(t)
 	req := httptest.NewRequest(http.MethodPost, "/api/v4/relay/pair?token=leak", strings.NewReader(`{}`))

internal/v4/payments/allocator.go

@@ -28,7 +28,7 @@ func NewAllocator() Allocator {
 	return Allocator{
 		Random:      cryptoRandomIndex,
 		SoftHorizon: defaultSoftHorizon,
-		Buckets:     2,
+		Buckets:     6,
 	}
 }
 
@@ -44,7 +44,7 @@ func (a Allocator) Select(ctx context.Context, tx *storage.ImmediateTx, profileI
 	}
 	buckets := a.Buckets
 	if buckets <= 0 {
-		buckets = 2
+		buckets = 6
 	}
 	randomIndex := a.Random
 	if randomIndex == nil {

internal/v4/payments/allocator_test.go

@@ -199,12 +199,14 @@ func TestAllocatorReleasesDueReservationBeforeChoosing(t *testing.T) {
 	}
 }
 
-func TestAllocatorReturnsCapacityOnlyWhenBothBucketsFull(t *testing.T) {
+func TestAllocatorReturnsCapacityOnlyWhenAllBucketsFull(t *testing.T) {
 	db := openAllocatorDB(t)
 	ctx := context.Background()
 	now := time.UnixMilli(1_788_200_000_000)
-	fillActiveRange(t, db.SQL, 10001, 10099, -1, now)
-	fillActiveRange(t, db.SQL, 10101, 10199, -1, now)
+	for bucket := int64(0); bucket < 6; bucket++ {
+		start := int64(10001) + bucket*100
+		fillActiveRange(t, db.SQL, start, start+98, -1, now)
+	}
 
 	allocator := NewAllocator()
 	err := db.WithImmediateTx(ctx, func(tx *storage.ImmediateTx) error {
@@ -216,6 +218,37 @@ func TestAllocatorReturnsCapacityOnlyWhenBothBucketsFull(t *testing.T) {
 	}
 }
 
+func TestAllocatorSupportsThreeHundredConcurrentPayments(t *testing.T) {
+	db := openAllocatorDB(t)
+	ctx := context.Background()
+	now := time.UnixMilli(1_788_200_000_000)
+	service := NewService(db)
+	service.Now = func() time.Time { return now }
+	service.Allocator.Random = fixedIndex(0)
+
+	seen := make(map[int64]struct{}, 300)
+	for i := 0; i < 300; i++ {
+		result, err := service.Create(ctx, CreateInput{
+			RequestedAmountPaise: 10000,
+			Name:                 fmt.Sprintf("Person %d", i),
+			ExternalID:           "evt_capacity_300",
+			Metadata:             []byte(`{}`),
+			IdempotencyScope:     "load-test",
+			IdempotencyKey:       fmt.Sprintf("payment-%03d", i),
+		})
+		if err != nil {
+			t.Fatalf("create %d failed: %v", i+1, err)
+		}
+		if _, exists := seen[result.Payment.PayableAmountPaise]; exists {
+			t.Fatalf("duplicate payable amount %d at create %d", result.Payment.PayableAmountPaise, i+1)
+		}
+		seen[result.Payment.PayableAmountPaise] = struct{}{}
+	}
+	if len(seen) != 300 {
+		t.Fatalf("unique payable amounts=%d, want 300", len(seen))
+	}
+}
+
 func TestAllocatorRejectsNonWholeRequestedAmount(t *testing.T) {
 	db := openAllocatorDB(t)
 	ctx := context.Background()
@@ -274,7 +307,7 @@ func insertTestPayment(t *testing.T, db *sql.DB, id string, payable int64, creat
 	t.Helper()
 	requested := int64(10000)
 	adjustment := payable - requested
-	if adjustment <= 0 || adjustment > 199 {
+	if adjustment <= 0 || adjustment > 599 {
 		t.Fatalf("invalid test payable %d", payable)
 	}
 	now := created.UnixMilli()

internal/v4/relay/service.go

@@ -23,7 +23,10 @@ import (
 const (
 	SchemaVersion            = 1
 	EventPath                = "/api/v4/relay/events"
+	EventBatchPath           = "/api/v4/relay/events/batch"
 	maxRawBodyBytes          = 64 << 10
+	maxBatchRawBodyBytes     = 1 << 20
+	maxBatchEvents           = 50
 	maxNotificationTextBytes = 32 << 10
 )
 
@@ -47,13 +50,23 @@ type EventInput struct {
 }
 
 type IngestResult struct {
+	EventID      string `json:"event_id,omitempty"`
 	RelayEventID string `json:"relay_event_id"`
 	Status       string `json:"status"`
 	PaymentID    string `json:"payment_id,omitempty"`
 	Duplicate    bool   `json:"duplicate,omitempty"`
 	Transitioned bool   `json:"transitioned,omitempty"`
 }
 
+type BatchInput struct {
+	SchemaVersion int               `json:"schema_version"`
+	Events        []json.RawMessage `json:"events"`
+}
+
+type BatchIngestResult struct {
+	Results []IngestResult `json:"results"`
+}
+
 func NewService(db *storage.DB, paymentService *payments.Service) *Service {
 	return &Service{
 		DB: db, Payments: paymentService, Now: time.Now, NewID: randomID,
@@ -168,9 +181,70 @@ func (s *Service) IngestSigned(ctx context.Context, auth RequestAuth, rawBody []
 	if err := validateEventInput(&input); err != nil {
 		return IngestResult{}, err
 	}
-	payloadHash := sha256.Sum256(rawBody)
+	return s.ingestVerifiedEvent(ctx, device, input, rawBody, now)
+}
+
+func (s *Service) IngestBatchSigned(ctx context.Context, auth RequestAuth, rawBody []byte) (BatchIngestResult, error) {
+	if s == nil || s.DB == nil || s.DB.SQL == nil || s.Payments == nil {
+		return BatchIngestResult{}, errors.New("relay storage and payment service are required")
+	}
+	if len(rawBody) == 0 || len(rawBody) > maxBatchRawBodyBytes {
+		return BatchIngestResult{}, relayError("RELAY_BATCH_TOO_LARGE", "relay batch body is empty or too large", 400)
+	}
+	if strings.ToUpper(strings.TrimSpace(auth.Method)) != "POST" || auth.Path != EventBatchPath {
+		return BatchIngestResult{}, relayError("INVALID_RELAY_ENDPOINT", "relay signature is not for the v4 batch endpoint", 401)
+	}
+	nowFn := s.Now
+	if nowFn == nil {
+		nowFn = time.Now
+	}
+	now := nowFn().UTC()
+	device, err := verifyRequest(ctx, s.DB, auth, rawBody, now)
+	if err != nil {
+		return BatchIngestResult{}, err
+	}
+	var batch BatchInput
+	if err := json.Unmarshal(rawBody, &batch); err != nil {
+		return BatchIngestResult{}, relayError("INVALID_RELAY_BATCH", "relay batch is not valid JSON", 400)
+	}
+	if batch.SchemaVersion != SchemaVersion {
+		return BatchIngestResult{}, relayError("UNSUPPORTED_RELAY_SCHEMA", "schema_version must be 1", 400)
+	}
+	if len(batch.Events) == 0 || len(batch.Events) > maxBatchEvents {
+		return BatchIngestResult{}, relayError("INVALID_RELAY_BATCH", "relay batch must contain between 1 and 50 events", 400)
+	}
+
+	inputs := make([]EventInput, len(batch.Events))
+	for i, rawEvent := range batch.Events {
+		if len(rawEvent) == 0 || len(rawEvent) > maxRawBodyBytes {
+			return BatchIngestResult{}, relayError("RELAY_EVENT_TOO_LARGE", "relay batch contains an empty or oversized event", 400)
+		}
+		if err := json.Unmarshal(rawEvent, &inputs[i]); err != nil {
+			return BatchIngestResult{}, relayError("INVALID_RELAY_EVENT", "relay batch contains invalid event JSON", 400)
+		}
+		if err := validateEventInput(&inputs[i]); err != nil {
+			return BatchIngestResult{}, err
+		}
+	}
+
+	out := BatchIngestResult{Results: make([]IngestResult, 0, len(inputs))}
+	for i, input := range inputs {
+		result, err := s.ingestVerifiedEvent(ctx, device, input, batch.Events[i], now)
+		if err != nil {
+			// Earlier items may already be durable. Retrying the entire signed batch
+			// is safe because source event IDs are idempotent per relay device.
+			return BatchIngestResult{}, err
+		}
+		out.Results = append(out.Results, result)
+	}
+	return out, nil
+}
+
+func (s *Service) ingestVerifiedEvent(ctx context.Context, device verifiedDevice, input EventInput, rawEvent []byte, now time.Time) (IngestResult, error) {
+	payloadHash := sha256.Sum256(rawEvent)
 	postedAt, postedReliable := sanitizePostedAt(input.PostedAtMS, now)
 	result, inserted, err := s.acceptEvent(ctx, device, input, postedAt, postedReliable, now, payloadHash[:])
+	result.EventID = input.EventID
 	if err != nil || !inserted {
 		return result, err
 	}
@@ -219,8 +293,6 @@ func (s *Service) IngestSigned(ctx context.Context, auth RequestAuth, rawBody []
 	}
 	matched, err := s.Payments.ApplyObservationForRelay(ctx, result.RelayEventID, obs, now, device.ID, device.EnrolledAt)
 	if errors.Is(err, payments.ErrRelayEventNotFound) {
-		// A retention worker or device revocation may have finalized this
-		// stale event while parsing was in flight. Do not apply its payload.
 		if finishErr := s.finishIgnored(ctx, result.RelayEventID, errors.New("relay event was no longer authorized for processing")); finishErr != nil {
 			return IngestResult{}, finishErr
 		}
@@ -364,7 +436,7 @@ func existingRelayEvent(ctx context.Context, tx *storage.ImmediateTx, deviceID,
 	result.PaymentID = paymentID.String
 	same := false
 	if len(storedHash) == sha256.Size {
-		same = bytes.Equal(storedHash, payloadHash)
+		same = bytes.Equal(storedHash, payloadHash) || relayEventFieldsEqual(packageName, storedPostedAt, amountHint, title, text, bigText, in, postedAt, postedReliable)
 	} else if bytes.HasPrefix(storedHash, []byte("legacy:")) {
 		comparisonPostedAt := postedAt
 		if !postedReliable {
@@ -374,15 +446,19 @@ func existingRelayEvent(ctx context.Context, tx *storage.ImmediateTx, deviceID,
 	} else if len(storedHash) == 0 {
 		// Legacy rows without a hash retain their fields until this service
 		// redacts them, so retries remain idempotent during migration.
-		same = packageName == in.PackageName &&
-			(!postedReliable || storedPostedAt == postedAt.UnixMilli()) &&
-			nullableAmountEqual(amountHint, in.AmountHintPaise) &&
-			nullableTextEqual(title, in.Title) &&
-			nullableTextEqual(text, in.Text) &&
-			nullableTextEqual(bigText, in.BigText)
+		same = relayEventFieldsEqual(packageName, storedPostedAt, amountHint, title, text, bigText, in, postedAt, postedReliable)
 	}
 	return result, true, same, nil
 }
+func relayEventFieldsEqual(packageName string, storedPostedAt int64, amountHint sql.NullInt64, title, text, bigText sql.NullString, in EventInput, postedAt time.Time, postedReliable bool) bool {
+	return packageName == in.PackageName &&
+		(!postedReliable || storedPostedAt == postedAt.UnixMilli()) &&
+		nullableAmountEqual(amountHint, in.AmountHintPaise) &&
+		nullableTextEqual(title, in.Title) &&
+		nullableTextEqual(text, in.Text) &&
+		nullableTextEqual(bigText, in.BigText)
+}
+
 func legacyPayloadFingerprint(packageName string, postedAt int64, amountHint sql.NullInt64,
 	title, text, bigText sql.NullString) []byte {
 	data := make([]byte, 0, 128)