AI Gateway Go 0.10.0 源码快照 + 旗舰版需求规划报告
M0-M7 已完成:核心网关(身份/RBAC/TOTP/OIDC/SAML/Provider/配额/路由/内容策略/审计/定价)+ 资源市场(MCP/Skills/数字员工)。 含 22 个 PostgreSQL 迁移、管理端/门户端前端源码、OpenAPI 契约、部署 compose。 Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,95 @@
|
||||
package outbox
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"net/http"
|
||||
"strconv"
|
||||
"strings"
|
||||
|
||||
"aigateway.local/core/internal/identity"
|
||||
"aigateway.local/core/internal/platform/apiresponse"
|
||||
"github.com/jackc/pgx/v5/pgtype"
|
||||
)
|
||||
|
||||
type AdminHTTPHandler struct {
|
||||
store *Store
|
||||
identity *identity.Service
|
||||
mux *http.ServeMux
|
||||
}
|
||||
|
||||
func NewAdminHTTPHandler(store *Store, identityService *identity.Service) *AdminHTTPHandler {
|
||||
h := &AdminHTTPHandler{store: store, identity: identityService, mux: http.NewServeMux()}
|
||||
h.mux.HandleFunc("GET /api/v1/admin/outbox-events", h.list)
|
||||
h.mux.HandleFunc("POST /api/v1/admin/outbox-events/{event_id}/retry", h.retry)
|
||||
return h
|
||||
}
|
||||
|
||||
func (h *AdminHTTPHandler) ServeHTTP(writer http.ResponseWriter, request *http.Request) {
|
||||
h.mux.ServeHTTP(writer, request)
|
||||
}
|
||||
|
||||
func (h *AdminHTTPHandler) list(writer http.ResponseWriter, request *http.Request) {
|
||||
if !h.requirePermission(writer, request, identity.PermissionOutboxRead) {
|
||||
return
|
||||
}
|
||||
status := strings.TrimSpace(request.URL.Query().Get("status"))
|
||||
if status != "" && status != "pending" && status != "dead" && status != "processed" {
|
||||
apiresponse.Error(writer, http.StatusBadRequest, "status 必须是 pending、dead 或 processed")
|
||||
return
|
||||
}
|
||||
limit := 100
|
||||
if value := request.URL.Query().Get("limit"); value != "" {
|
||||
parsed, err := strconv.Atoi(value)
|
||||
if err != nil || parsed < 1 || parsed > 500 {
|
||||
apiresponse.Error(writer, http.StatusBadRequest, "limit 必须在 1 到 500 之间")
|
||||
return
|
||||
}
|
||||
limit = parsed
|
||||
}
|
||||
items, err := h.store.List(request.Context(), status, strings.TrimSpace(request.URL.Query().Get("type")), limit)
|
||||
if err != nil {
|
||||
apiresponse.Error(writer, http.StatusServiceUnavailable, "事件投递查询服务暂不可用")
|
||||
return
|
||||
}
|
||||
apiresponse.OK(writer, items)
|
||||
}
|
||||
|
||||
func (h *AdminHTTPHandler) retry(writer http.ResponseWriter, request *http.Request) {
|
||||
if !h.requirePermission(writer, request, identity.PermissionOutboxManage) {
|
||||
return
|
||||
}
|
||||
eventID := strings.TrimSpace(request.PathValue("event_id"))
|
||||
var parsedID pgtype.UUID
|
||||
if err := parsedID.Scan(eventID); err != nil || !parsedID.Valid {
|
||||
apiresponse.Error(writer, http.StatusBadRequest, "event_id 必须是有效 UUID")
|
||||
return
|
||||
}
|
||||
if err := h.store.Retry(request.Context(), eventID); err != nil {
|
||||
if errors.Is(err, ErrEventNotFound) {
|
||||
apiresponse.Error(writer, http.StatusNotFound, "待处理或死信事件不存在")
|
||||
return
|
||||
}
|
||||
apiresponse.Error(writer, http.StatusServiceUnavailable, "事件重试服务暂不可用")
|
||||
return
|
||||
}
|
||||
apiresponse.OK(writer, map[string]bool{"retried": true})
|
||||
}
|
||||
|
||||
func (h *AdminHTTPHandler) requirePermission(writer http.ResponseWriter, request *http.Request, permission string) bool {
|
||||
account, err := h.identity.Authenticate(request.Context(), identity.KindAdmin, request.Header.Get("Authorization"))
|
||||
if err != nil {
|
||||
if errors.Is(err, identity.ErrInvalidSession) || errors.Is(err, identity.ErrNotFound) {
|
||||
apiresponse.Error(writer, http.StatusUnauthorized, "登录状态无效或已过期")
|
||||
} else if errors.Is(err, identity.ErrAccountDisabled) {
|
||||
apiresponse.Error(writer, http.StatusForbidden, "管理员账号已被停用")
|
||||
} else {
|
||||
apiresponse.Error(writer, http.StatusServiceUnavailable, "身份服务暂不可用")
|
||||
}
|
||||
return false
|
||||
}
|
||||
if !identity.HasPermission(account, permission) {
|
||||
apiresponse.Error(writer, http.StatusForbidden, "缺少事件投递操作权限")
|
||||
return false
|
||||
}
|
||||
return true
|
||||
}
|
||||
@@ -0,0 +1,55 @@
|
||||
package outbox
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"os"
|
||||
"testing"
|
||||
|
||||
"github.com/jackc/pgx/v5"
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
)
|
||||
|
||||
func TestConsumeTransactionRollbackAndIdempotency(t *testing.T) {
|
||||
databaseURL := os.Getenv("OUTBOX_TEST_DATABASE_URL")
|
||||
if databaseURL == "" {
|
||||
t.Skip("OUTBOX_TEST_DATABASE_URL is not configured")
|
||||
}
|
||||
ctx := context.Background()
|
||||
pool, err := pgxpool.New(ctx, databaseURL)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer pool.Close()
|
||||
store := NewStore(pool)
|
||||
const subscriber = "outbox-integration-test"
|
||||
const eventID = "33333333-3333-4333-8333-333333333333"
|
||||
_, _ = pool.Exec(ctx, `DELETE FROM gateway.event_consumptions WHERE subscriber=$1 AND event_id=$2`, subscriber, eventID)
|
||||
t.Cleanup(func() {
|
||||
_, _ = pool.Exec(context.Background(), `DELETE FROM gateway.event_consumptions WHERE subscriber=$1 AND event_id=$2`, subscriber, eventID)
|
||||
})
|
||||
|
||||
calls := 0
|
||||
wantErr := errors.New("rollback handler")
|
||||
consumed, err := store.Consume(ctx, subscriber, eventID, func(context.Context, pgx.Tx) error {
|
||||
calls++
|
||||
return wantErr
|
||||
})
|
||||
if consumed || !errors.Is(err, wantErr) {
|
||||
t.Fatalf("failed handler must roll back: consumed=%v err=%v", consumed, err)
|
||||
}
|
||||
consumed, err = store.Consume(ctx, subscriber, eventID, func(context.Context, pgx.Tx) error {
|
||||
calls++
|
||||
return nil
|
||||
})
|
||||
if !consumed || err != nil {
|
||||
t.Fatalf("second attempt must consume: consumed=%v err=%v", consumed, err)
|
||||
}
|
||||
consumed, err = store.Consume(ctx, subscriber, eventID, func(context.Context, pgx.Tx) error {
|
||||
calls++
|
||||
return nil
|
||||
})
|
||||
if consumed || err != nil || calls != 2 {
|
||||
t.Fatalf("duplicate must skip handler: consumed=%v calls=%d err=%v", consumed, calls, err)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,98 @@
|
||||
package outbox
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"strconv"
|
||||
"time"
|
||||
|
||||
"github.com/redis/go-redis/v9"
|
||||
)
|
||||
|
||||
var ErrPublisherUnavailable = errors.New("outbox publisher unavailable")
|
||||
|
||||
type PublishResult struct {
|
||||
Published bool
|
||||
StreamID string
|
||||
}
|
||||
|
||||
type RedisPublisher struct {
|
||||
client *redis.Client
|
||||
stream string
|
||||
maxLength int64
|
||||
markerTTL time.Duration
|
||||
script *redis.Script
|
||||
}
|
||||
|
||||
func NewRedisPublisher(client *redis.Client, stream string, maxLength int64, markerTTL time.Duration) *RedisPublisher {
|
||||
return &RedisPublisher{client: client, stream: stream, maxLength: maxLength, markerTTL: markerTTL, script: redis.NewScript(publishScript)}
|
||||
}
|
||||
|
||||
// Ping reports whether the downstream Redis stream is reachable. The outbox
|
||||
// worker calls it before claiming events so a Redis outage never consumes the
|
||||
// events' delivery budget (claiming increments attempts; dead-lettering then
|
||||
// burns the whole queue for a fault that was never the events').
|
||||
func (p *RedisPublisher) Ping(ctx context.Context) error {
|
||||
if p == nil || p.client == nil {
|
||||
return ErrPublisherUnavailable
|
||||
}
|
||||
if err := p.client.Ping(ctx).Err(); err != nil {
|
||||
return fmt.Errorf("%w: %v", ErrPublisherUnavailable, err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (p *RedisPublisher) Publish(ctx context.Context, event Event) (PublishResult, error) {
|
||||
if p == nil || p.client == nil {
|
||||
return PublishResult{}, ErrPublisherUnavailable
|
||||
}
|
||||
marker := "gateway:{outbox}:published:" + event.EventID
|
||||
result, err := p.script.Run(ctx, p.client, []string{marker, p.stream},
|
||||
int64(p.markerTTL.Seconds()), p.maxLength, event.EventID, event.EventType, event.EventVersion,
|
||||
valueOrEmpty(event.TenantID), event.AggregateType, event.AggregateID, string(event.Payload), string(event.TraceContext), event.OccurredAt.UTC().Format(time.RFC3339Nano),
|
||||
).Slice()
|
||||
if err != nil || len(result) != 2 {
|
||||
return PublishResult{}, fmt.Errorf("%w: %v", ErrPublisherUnavailable, err)
|
||||
}
|
||||
published, err := redisInt(result[0])
|
||||
if err != nil {
|
||||
return PublishResult{}, fmt.Errorf("%w: %v", ErrPublisherUnavailable, err)
|
||||
}
|
||||
streamID := "duplicate"
|
||||
if result[1] != nil && fmt.Sprint(result[1]) != "" {
|
||||
streamID = fmt.Sprint(result[1])
|
||||
}
|
||||
return PublishResult{Published: published == 1, StreamID: streamID}, nil
|
||||
}
|
||||
|
||||
func valueOrEmpty(value *string) string {
|
||||
if value == nil {
|
||||
return ""
|
||||
}
|
||||
return *value
|
||||
}
|
||||
|
||||
func redisInt(value any) (int64, error) {
|
||||
switch typed := value.(type) {
|
||||
case int64:
|
||||
return typed, nil
|
||||
case string:
|
||||
return strconv.ParseInt(typed, 10, 64)
|
||||
case []byte:
|
||||
return strconv.ParseInt(string(typed), 10, 64)
|
||||
default:
|
||||
return 0, fmt.Errorf("unexpected redis integer %T", value)
|
||||
}
|
||||
}
|
||||
|
||||
const publishScript = `
|
||||
if redis.call('SET', KEYS[1], '1', 'NX', 'EX', tonumber(ARGV[1])) then
|
||||
local id = redis.call('XADD', KEYS[2], 'MAXLEN', '~', tonumber(ARGV[2]), '*',
|
||||
'event_id', ARGV[3], 'event_type', ARGV[4], 'event_version', ARGV[5],
|
||||
'tenant_id', ARGV[6], 'aggregate_type', ARGV[7], 'aggregate_id', ARGV[8],
|
||||
'payload', ARGV[9], 'trace_context', ARGV[10], 'occurred_at', ARGV[11])
|
||||
return {1, id}
|
||||
end
|
||||
return {0, ''}
|
||||
`
|
||||
@@ -0,0 +1,201 @@
|
||||
package outbox
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/jackc/pgx/v5"
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
)
|
||||
|
||||
var (
|
||||
ErrStoreUnavailable = errors.New("outbox store unavailable")
|
||||
ErrEventNotFound = errors.New("outbox event not found")
|
||||
)
|
||||
|
||||
type Event struct {
|
||||
EventID string `json:"event_id"`
|
||||
EventType string `json:"event_type"`
|
||||
EventVersion int `json:"event_version"`
|
||||
TenantID *string `json:"tenant_id"`
|
||||
AggregateType string `json:"aggregate_type"`
|
||||
AggregateID string `json:"aggregate_id"`
|
||||
Payload json.RawMessage `json:"payload"`
|
||||
TraceContext json.RawMessage `json:"trace_context"`
|
||||
OccurredAt time.Time `json:"occurred_at"`
|
||||
Attempts int `json:"attempts"`
|
||||
}
|
||||
|
||||
type Store struct{ pool *pgxpool.Pool }
|
||||
|
||||
func NewStore(pool *pgxpool.Pool) *Store { return &Store{pool: pool} }
|
||||
|
||||
func (s *Store) Claim(ctx context.Context, workerID string, limit int, lease time.Duration) ([]Event, error) {
|
||||
if s == nil || s.pool == nil {
|
||||
return nil, ErrStoreUnavailable
|
||||
}
|
||||
rows, err := s.pool.Query(ctx, `
|
||||
WITH candidates AS (
|
||||
SELECT event_id FROM gateway.outbox_events
|
||||
WHERE processed_at IS NULL AND dead_lettered_at IS NULL AND available_at <= clock_timestamp()
|
||||
AND (locked_at IS NULL OR locked_at < clock_timestamp()-($1 * interval '1 second'))
|
||||
ORDER BY available_at,occurred_at
|
||||
FOR UPDATE SKIP LOCKED LIMIT $2
|
||||
)
|
||||
UPDATE gateway.outbox_events e
|
||||
SET locked_at=clock_timestamp(),locked_by=$3,attempts=e.attempts+1
|
||||
FROM candidates c WHERE e.event_id=c.event_id
|
||||
RETURNING e.event_id::text,e.event_type,e.event_version,e.tenant_id::text,e.aggregate_type,
|
||||
e.aggregate_id,e.payload,e.trace_context,e.occurred_at,e.attempts`, lease.Seconds(), limit, workerID)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("%w: claim: %v", ErrStoreUnavailable, err)
|
||||
}
|
||||
defer rows.Close()
|
||||
events := make([]Event, 0, limit)
|
||||
for rows.Next() {
|
||||
var event Event
|
||||
if err := rows.Scan(&event.EventID, &event.EventType, &event.EventVersion, &event.TenantID,
|
||||
&event.AggregateType, &event.AggregateID, &event.Payload, &event.TraceContext, &event.OccurredAt, &event.Attempts); err != nil {
|
||||
return nil, fmt.Errorf("%w: scan claim: %v", ErrStoreUnavailable, err)
|
||||
}
|
||||
events = append(events, event)
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
return nil, fmt.Errorf("%w: claim rows: %v", ErrStoreUnavailable, err)
|
||||
}
|
||||
return events, nil
|
||||
}
|
||||
|
||||
func (s *Store) MarkProcessed(ctx context.Context, eventID, workerID, streamID string) error {
|
||||
if s == nil || s.pool == nil {
|
||||
return ErrStoreUnavailable
|
||||
}
|
||||
result, err := s.pool.Exec(ctx, `UPDATE gateway.outbox_events SET processed_at=clock_timestamp(),locked_at=NULL,locked_by=NULL,last_error=NULL,published_stream_id=$3 WHERE event_id=$1 AND locked_by=$2 AND processed_at IS NULL`, eventID, workerID, streamID)
|
||||
if err != nil {
|
||||
return fmt.Errorf("%w: mark processed: %v", ErrStoreUnavailable, err)
|
||||
}
|
||||
if result.RowsAffected() != 1 {
|
||||
return ErrEventNotFound
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *Store) MarkFailed(ctx context.Context, event Event, workerID string, deliveryErr error, maxAttempts int, delay time.Duration) error {
|
||||
if s == nil || s.pool == nil {
|
||||
return ErrStoreUnavailable
|
||||
}
|
||||
message := deliveryErr.Error()
|
||||
if len(message) > 2048 {
|
||||
message = message[:2048]
|
||||
}
|
||||
dead := event.Attempts >= maxAttempts
|
||||
result, err := s.pool.Exec(ctx, `
|
||||
UPDATE gateway.outbox_events SET locked_at=NULL,locked_by=NULL,last_error=$3,
|
||||
available_at=CASE WHEN $4 THEN available_at ELSE clock_timestamp()+($5 * interval '1 second') END,
|
||||
dead_lettered_at=CASE WHEN $4 THEN clock_timestamp() ELSE NULL END
|
||||
WHERE event_id=$1 AND locked_by=$2 AND processed_at IS NULL`, event.EventID, workerID, message, dead, delay.Seconds())
|
||||
if err != nil {
|
||||
return fmt.Errorf("%w: mark failed: %v", ErrStoreUnavailable, err)
|
||||
}
|
||||
if result.RowsAffected() != 1 {
|
||||
return ErrEventNotFound
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
type EventView struct {
|
||||
Event
|
||||
AvailableAt time.Time `json:"available_at"`
|
||||
LockedAt *time.Time `json:"locked_at"`
|
||||
LockedBy *string `json:"locked_by"`
|
||||
ProcessedAt *time.Time `json:"processed_at"`
|
||||
DeadLetteredAt *time.Time `json:"dead_lettered_at"`
|
||||
LastError *string `json:"last_error"`
|
||||
PublishedStream *string `json:"published_stream_id"`
|
||||
}
|
||||
|
||||
func (s *Store) List(ctx context.Context, status, eventType string, limit int) ([]EventView, error) {
|
||||
if s == nil || s.pool == nil {
|
||||
return nil, ErrStoreUnavailable
|
||||
}
|
||||
where := "TRUE"
|
||||
switch status {
|
||||
case "pending":
|
||||
where = "processed_at IS NULL AND dead_lettered_at IS NULL"
|
||||
case "dead":
|
||||
where = "dead_lettered_at IS NOT NULL"
|
||||
case "processed":
|
||||
where = "processed_at IS NOT NULL"
|
||||
}
|
||||
args := []any{limit}
|
||||
if eventType != "" {
|
||||
args = append(args, eventType)
|
||||
where += fmt.Sprintf(" AND event_type=$%d", len(args))
|
||||
}
|
||||
rows, err := s.pool.Query(ctx, `SELECT event_id::text,event_type,event_version,tenant_id::text,aggregate_type,aggregate_id,payload,trace_context,occurred_at,attempts,available_at,locked_at,locked_by,processed_at,dead_lettered_at,last_error,published_stream_id FROM gateway.outbox_events WHERE `+where+` ORDER BY occurred_at DESC LIMIT $1`, args...)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("%w: list: %v", ErrStoreUnavailable, err)
|
||||
}
|
||||
defer rows.Close()
|
||||
items := make([]EventView, 0)
|
||||
for rows.Next() {
|
||||
var item EventView
|
||||
if err := rows.Scan(&item.EventID, &item.EventType, &item.EventVersion, &item.TenantID, &item.AggregateType,
|
||||
&item.AggregateID, &item.Payload, &item.TraceContext, &item.OccurredAt, &item.Attempts, &item.AvailableAt,
|
||||
&item.LockedAt, &item.LockedBy, &item.ProcessedAt, &item.DeadLetteredAt, &item.LastError, &item.PublishedStream); err != nil {
|
||||
return nil, fmt.Errorf("%w: scan list: %v", ErrStoreUnavailable, err)
|
||||
}
|
||||
items = append(items, item)
|
||||
}
|
||||
return items, rows.Err()
|
||||
}
|
||||
|
||||
func (s *Store) Retry(ctx context.Context, eventID string) error {
|
||||
if s == nil || s.pool == nil {
|
||||
return ErrStoreUnavailable
|
||||
}
|
||||
result, err := s.pool.Exec(ctx, `UPDATE gateway.outbox_events SET attempts=0,available_at=clock_timestamp(),locked_at=NULL,locked_by=NULL,dead_lettered_at=NULL,last_error=NULL WHERE event_id=$1 AND processed_at IS NULL`, eventID)
|
||||
if err != nil {
|
||||
return fmt.Errorf("%w: retry: %v", ErrStoreUnavailable, err)
|
||||
}
|
||||
if result.RowsAffected() != 1 {
|
||||
return ErrEventNotFound
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
type ConsumerHandler func(context.Context, pgx.Tx) error
|
||||
|
||||
func (s *Store) Consume(ctx context.Context, subscriber, eventID string, handler ConsumerHandler) (bool, error) {
|
||||
if s == nil || s.pool == nil {
|
||||
return false, ErrStoreUnavailable
|
||||
}
|
||||
subscriber = strings.TrimSpace(subscriber)
|
||||
if subscriber == "" || eventID == "" || handler == nil {
|
||||
return false, errors.New("subscriber, event ID, and handler are required")
|
||||
}
|
||||
tx, err := s.pool.Begin(ctx)
|
||||
if err != nil {
|
||||
return false, fmt.Errorf("%w: begin consumption: %v", ErrStoreUnavailable, err)
|
||||
}
|
||||
defer func() { _ = tx.Rollback(ctx) }()
|
||||
var inserted bool
|
||||
err = tx.QueryRow(ctx, `WITH inserted AS (INSERT INTO gateway.event_consumptions(subscriber,event_id) VALUES($1,$2) ON CONFLICT DO NOTHING RETURNING 1) SELECT EXISTS(SELECT 1 FROM inserted)`, subscriber, eventID).Scan(&inserted)
|
||||
if err != nil {
|
||||
return false, fmt.Errorf("%w: reserve consumption: %v", ErrStoreUnavailable, err)
|
||||
}
|
||||
if !inserted {
|
||||
return false, nil
|
||||
}
|
||||
if err := handler(ctx, tx); err != nil {
|
||||
return false, err
|
||||
}
|
||||
if err := tx.Commit(ctx); err != nil {
|
||||
return false, fmt.Errorf("%w: commit consumption: %v", ErrStoreUnavailable, err)
|
||||
}
|
||||
return true, nil
|
||||
}
|
||||
@@ -0,0 +1,91 @@
|
||||
package outbox
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"log/slog"
|
||||
"math"
|
||||
"time"
|
||||
)
|
||||
|
||||
type WorkerConfig struct {
|
||||
WorkerID string
|
||||
BatchSize int
|
||||
PollInterval time.Duration
|
||||
Lease time.Duration
|
||||
MaxAttempts int
|
||||
MaxBackoff time.Duration
|
||||
}
|
||||
|
||||
type Worker struct {
|
||||
store *Store
|
||||
publisher *RedisPublisher
|
||||
config WorkerConfig
|
||||
logger *slog.Logger
|
||||
}
|
||||
|
||||
func NewWorker(store *Store, publisher *RedisPublisher, config WorkerConfig, logger *slog.Logger) *Worker {
|
||||
return &Worker{store: store, publisher: publisher, config: config, logger: logger}
|
||||
}
|
||||
|
||||
func (w *Worker) Run(ctx context.Context) error {
|
||||
for {
|
||||
processed, err := w.runBatch(ctx)
|
||||
if err != nil && ctx.Err() == nil && w.logger != nil {
|
||||
w.logger.Error("outbox batch failed", "error", err)
|
||||
}
|
||||
if ctx.Err() != nil {
|
||||
return nil
|
||||
}
|
||||
if err == nil && processed > 0 {
|
||||
continue
|
||||
}
|
||||
timer := time.NewTimer(w.config.PollInterval)
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
timer.Stop()
|
||||
return nil
|
||||
case <-timer.C:
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (w *Worker) runBatch(ctx context.Context) (int, error) {
|
||||
// Gate on the downstream publisher before claiming anything. Claiming
|
||||
// increments each event's attempt counter, so claiming during a Redis
|
||||
// outage would burn every queued event's delivery budget and dead-letter
|
||||
// the whole queue the moment the budget ran out — even though the events
|
||||
// themselves were never at fault. When Redis is unreachable we back off
|
||||
// instead, leaving events untouched in PostgreSQL until it recovers.
|
||||
if err := w.publisher.Ping(ctx); err != nil {
|
||||
return 0, err
|
||||
}
|
||||
events, err := w.store.Claim(ctx, w.config.WorkerID, w.config.BatchSize, w.config.Lease)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
var batchErr error
|
||||
for _, event := range events {
|
||||
result, publishErr := w.publisher.Publish(ctx, event)
|
||||
if publishErr == nil {
|
||||
if err := w.store.MarkProcessed(ctx, event.EventID, w.config.WorkerID, result.StreamID); err != nil {
|
||||
batchErr = errors.Join(batchErr, err)
|
||||
}
|
||||
continue
|
||||
}
|
||||
delay := retryDelay(event.Attempts, w.config.MaxBackoff)
|
||||
if err := w.store.MarkFailed(ctx, event, w.config.WorkerID, publishErr, w.config.MaxAttempts, delay); err != nil {
|
||||
batchErr = errors.Join(batchErr, err)
|
||||
}
|
||||
}
|
||||
return len(events), batchErr
|
||||
}
|
||||
|
||||
func retryDelay(attempt int, maximum time.Duration) time.Duration {
|
||||
seconds := math.Pow(2, float64(max(attempt-1, 0)))
|
||||
delay := time.Duration(seconds * float64(time.Second))
|
||||
if delay > maximum {
|
||||
return maximum
|
||||
}
|
||||
return delay
|
||||
}
|
||||
@@ -0,0 +1,18 @@
|
||||
package outbox
|
||||
|
||||
import (
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func TestRetryDelayIsExponentialAndCapped(t *testing.T) {
|
||||
tests := []struct {
|
||||
attempt int
|
||||
want time.Duration
|
||||
}{{1, time.Second}, {2, 2 * time.Second}, {3, 4 * time.Second}, {20, 5 * time.Second}}
|
||||
for _, test := range tests {
|
||||
if got := retryDelay(test.attempt, 5*time.Second); got != test.want {
|
||||
t.Fatalf("attempt %d: got %s, want %s", test.attempt, got, test.want)
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user