Replay Strategy #

// ANTI-PATTERN: unlimited retry, no delay — bombarding the downstream
func processEvent(event Event) error {
    for {
        err := handler.Process(event)
        if err == nil {
            return nil
        }
        // if the downstream is down, this loops forever and makes things worse
    }
}

// CORRECT: retry with a clear maximum
func processWithRetry(event Event, maxAttempts int) error {
    var lastErr error
    for attempt := 1; attempt <= maxAttempts; attempt++ {
        lastErr = handler.Process(event)
        if lastErr == nil {
            return nil
        }
        log.Warnf("attempt %d/%d failed for event %s: %v",
            attempt, maxAttempts, event.ID, lastErr)
    }
    return fmt.Errorf("all %d attempts failed: %w", maxAttempts, lastErr)
}

// ANTI-PATTERN: fixed delay — not adaptive to how long the failure lasts
func processWithFixedRetry(event Event) error {
    for i := 0; i < 5; i++ {
        if err := handler.Process(event); err == nil {
            return nil
        }
        time.Sleep(1 * time.Second) // always 1 second, no matter how long it's down
    }
    return errors.New("max retries exceeded")
}

// CORRECT: exponential backoff with jitter prevents thundering herds
func processWithBackoff(ctx context.Context, event Event) error {
    baseDelay := 1 * time.Second
    maxDelay  := 5 * time.Minute
    maxAttempts := 8

    for attempt := 1; attempt <= maxAttempts; attempt++ {
        err := handler.Process(event)
        if err == nil {
            return nil
        }

        if attempt == maxAttempts {
            return fmt.Errorf("exhausted %d attempts: %w", maxAttempts, err)
        }

        // Delay: 1s, 2s, 4s, 8s, 16s, 32s, 60s (capped)...
        delay := baseDelay * time.Duration(1<<uint(attempt-1))
        if delay > maxDelay {
            delay = maxDelay
        }

        // Jitter ±25%: so not all consumers retry at exactly the same time
        jitter := time.Duration(rand.Int63n(int64(delay / 4)))
        if rand.Intn(2) == 0 {
            delay += jitter
        } else {
            delay -= jitter
        }

        log.Infof("retry attempt %d in %v for event %s", attempt+1, delay, event.ID)

        select {
        case <-ctx.Done():
            return ctx.Err()
        case <-time.After(delay):
        }
    }
    return nil
}

// ANTI-PATTERN: dropping the event on failure — losing data
func (c *Consumer) handleMessage(msg Message) error {
    if err := c.process(msg); err != nil {
        log.Errorf("failed to process %s, dropping: %v", msg.ID, err)
        return nil // event lost forever
    }
    return nil
}

// CORRECT: send to the DLQ with full metadata, don't drop the event
func (c *Consumer) handleWithDLQ(ctx context.Context, msg Message) error {
    err := c.processWithBackoff(ctx, msg)
    if err == nil {
        return nil
    }

    // Enrich with metadata for later debugging
    dlqMsg := DLQMessage{
        OriginalMessage: msg,
        FailureReason:   err.Error(),
        AttemptCount:    msg.ApproximateReceiveCount,
        FailedAt:        time.Now(),
        ConsumerVersion: c.version,
        StackTrace:      debug.Stack(),
    }

    if dlqErr := c.dlqQueue.Send(ctx, dlqMsg); dlqErr != nil {
        // DLQ send failed too — log CRITICAL, needs immediate attention
        log.Criticalf("FAILED to send to DLQ, event %s will be lost: %v",
            msg.ID, dlqErr)
        return dlqErr
    }

    log.Warnf("event %s sent to DLQ after %d attempts: %v",
        msg.ID, msg.ApproximateReceiveCount, err)
    return nil // ack the main message so it doesn't loop in the main queue
}

// ANTI-PATTERN: replay without throttling — flooding the downstream all at once
func replayAllFromDLQ() {
    events := dlq.GetAll() // could be millions of events
    for _, e := range events {
        process(e) // all processed as fast as possible — the system could go down
    }
}

// CORRECT: controlled replay with a rate limiter and progress tracking
func (r *ReplayService) ReplayFromDLQ(ctx context.Context, opts ReplayOptions) error {
    log.Infof("starting replay: from=%v to=%v filter=%s rate=%f/s dryRun=%v",
        opts.From, opts.To, opts.Filter, opts.RatePerSecond, opts.DryRun)

    events, err := r.dlqStore.Query(ctx, opts.From, opts.To, opts.Filter)
    if err != nil {
        return fmt.Errorf("query DLQ: %w", err)
    }

    log.Infof("found %d events to replay", len(events))

    limiter := rate.NewLimiter(rate.Limit(opts.RatePerSecond), 1)
    var succeeded, failed, skipped int

    for i, event := range events {
        // Can be cancelled at any time via context
        if err := limiter.Wait(ctx); err != nil {
            log.Infof("replay cancelled at event %d/%d", i, len(events))
            return err
        }

        if opts.DryRun {
            log.Infof("[DRY RUN] would replay event %s", event.ID)
            skipped++
            continue
        }

        if err := r.processor.Process(ctx, event); err != nil {
            log.Errorf("replay failed for event %s: %v", event.ID, err)
            failed++
            continue
        }
        succeeded++

        // Progress log every 100 events
        if succeeded%100 == 0 {
            log.Infof("replay progress: %d/%d (success=%d fail=%d skip=%d)",
                i+1, len(events), succeeded, failed, skipped)
        }
    }

    log.Infof("replay complete: total=%d succeeded=%d failed=%d skipped=%d",
        len(events), succeeded, failed, skipped)
    return nil
}

// ANTI-PATTERN: rebuilding directly in the production table without a backup
func rebuildProjection() {
    db.Exec("DELETE FROM order_summary") // dangerous! no rollback if it fails
    for _, event := range eventLog.GetAll() {
        applyEvent(event)
    }
}

// CORRECT: rebuild into a shadow table, swap atomically when done
func (p *ProjectionRebuilder) RebuildOrderProjection(ctx context.Context) error {
    shadowTable := fmt.Sprintf("order_summary_rebuild_%d", time.Now().Unix())

    // Create the shadow table first
    if err := p.db.CreateTableLike(shadowTable, "order_summary"); err != nil {
        return fmt.Errorf("create shadow table: %w", err)
    }

    // Stream all events from the start into the shadow table
    events, err := p.eventLog.StreamAll(ctx, "orders", StreamOptions{
        FromOffset: 0,
        BatchSize:  500,
    })
    if err != nil {
        p.db.DropTable(shadowTable) // cleanup
        return fmt.Errorf("stream events: %w", err)
    }

    var count int
    for event := range events {
        if err := p.applyEventToTable(ctx, shadowTable, event); err != nil {
            p.db.DropTable(shadowTable)
            return fmt.Errorf("apply event %d: %w", event.Sequence, err)
        }
        count++
        if count%1000 == 0 {
            log.Infof("rebuilt %d events...", count)
        }
    }

    // Atomic swap: rename shadow → production
    if err := p.db.RenameTable("order_summary", "order_summary_old"); err != nil {
        return err
    }
    if err := p.db.RenameTable(shadowTable, "order_summary"); err != nil {
        return err
    }
    p.db.DropTable("order_summary_old")

    log.Infof("projection rebuild complete: %d events applied", count)
    return nil
}

// ANTI-PATTERN: non-idempotent consumer — replay is a disaster
func (c *Consumer) handlePayment(event PaymentRequestedEvent) error {
    // If this event is replayed, the payment is processed twice → double charge
    return c.paymentGateway.Charge(event.UserID, event.Amount)
}

// CORRECT: idempotent consumer — check event_id before processing
func (c *Consumer) handlePayment(ctx context.Context, event PaymentRequestedEvent) error {
    processed, err := c.processedEvents.Exists(ctx, event.EventID)
    if err != nil {
        return fmt.Errorf("check processed: %w", err)
    }
    if processed {
        log.Infof("event %s already processed, skipping", event.EventID)
        return nil // safe to skip during replay
    }

    // Process in a transaction: charge + mark as processed — atomic
    return c.db.Transaction(func(tx *gorm.DB) error {
        if err := c.paymentGateway.Charge(event.UserID, event.Amount); err != nil {
            return err
        }
        return tx.Create(&ProcessedEvent{
            EventID:     event.EventID,
            ProcessedAt: time.Now(),
        }).Error
    })
}

// ANTI-PATTERN: publishing without a partition key — order isn't guaranteed across partitions
func (p *Producer) publishOrderEvent(event OrderEvent) error {
    msg := &sarama.ProducerMessage{
        Topic: "order-events",
        // Without a key: Kafka distributes to partitions round-robin
        // OrderCreated (partition 0) could be processed after OrderShipped (partition 1)
        Value: sarama.ByteEncoder(mustMarshal(event)),
    }
    _, _, err := p.client.SendMessage(msg)
    return err
}

// CORRECT: use the entity ID as the partition key
// All events for the same order are guaranteed to land in the same partition
func (p *Producer) publishOrderEvent(event OrderEvent) error {
    msg := &sarama.ProducerMessage{
        Topic: "order-events",
        Key:   sarama.StringEncoder(event.OrderID), // ← the ordering key
        Value: sarama.ByteEncoder(mustMarshal(event)),
    }
    _, _, err := p.client.SendMessage(msg)
    return err
}

// ANTI-PATTERN: side effects directly in the handler without a state check
func (h *OrderHandler) handleOrderConfirmed(event OrderConfirmedEvent) error {
    h.updateDatabase(event)
    h.emailService.SendConfirmation(event.UserEmail) // resent on every replay!
    h.webhookService.Notify(event)                   // called again on every replay!
    return nil
}

// CORRECT: check state first — side effects only fire if the state hasn't changed
func (h *OrderHandler) handleOrderConfirmed(ctx context.Context, event OrderConfirmedEvent) error {
    order, err := h.orderRepo.FindByID(event.OrderID)
    if err != nil {
        return err
    }

    if order.Status == "CONFIRMED" {
        // State is already correct — all side effects already ran before
        // Replay is safe: no duplicate emails, no duplicate webhooks
        return nil
    }

    // Update the state first
    if err := h.orderRepo.UpdateStatus(event.OrderID, "CONFIRMED"); err != nil {
        return err
    }

    // Side effects only execute once because the state has already changed
    h.emailService.SendConfirmation(event.UserEmail)
    h.webhookService.Notify(event)
    return nil
}

// ANTI-PATTERN: replaying the entire DLQ at once — flooding the system
func replayAll() {
    events := dlq.GetAll() // could be millions of events
    for _, e := range events {
        process(e) // consumes all downstream resources at once
    }
}

// CORRECT: throttle + dry run + kill switch via context
type ReplayConfig struct {
    RatePerSecond float64       // how many events per second to process
    BatchSize     int           // how many events per batch
    DryRun        bool          // simulation without real execution
    From          time.Time     // filter from a certain time
    To            time.Time     // filter up to a certain time
    Filter        string        // filter by event type or other criteria
}

func (r *ReplayService) ReplayWithControl(ctx context.Context, cfg ReplayConfig) error {
    if cfg.DryRun {
        log.Info("[DRY RUN] no events will actually be processed")
    }
    limiter := rate.NewLimiter(rate.Limit(cfg.RatePerSecond), cfg.BatchSize)

    for _, event := range r.fetchEvents(cfg) {
        // Kill switch: ctx.Cancel() from outside stops the replay at any time
        if err := limiter.Wait(ctx); err != nil {
            log.Info("replay stopped by cancellation")
            return err
        }
        if !cfg.DryRun {
            r.processor.Process(ctx, event)
        }
    }
    return nil
}

// CORRECT: informative logs and metrics for every replay operation
type ReplayMetrics struct {
    TotalEvents  int
    Succeeded    int
    Failed       int
    Skipped      int           // idempotency skip
    Duration     time.Duration
    TriggeredBy  string        // who triggered it: "alerting-system", "engineer-unis"
    ReplayReason string        // why: "bug-fix-deploy-v2.3.1", "db-migration"
}

func (r *ReplayService) logCompletion(m ReplayMetrics) {
    log.Infof("[REPLAY COMPLETE] reason=%q triggered_by=%q "+
        "total=%d succeeded=%d failed=%d skipped=%d duration=%v",
        m.ReplayReason, m.TriggeredBy,
        m.TotalEvents, m.Succeeded, m.Failed, m.Skipped, m.Duration)

    // Send to the metrics system for dashboards and alerting
    metrics.Gauge("replay.success_rate",
        float64(m.Succeeded)/float64(m.TotalEvents)*100)
    metrics.Histogram("replay.duration_seconds", m.Duration.Seconds())

    // Alert if the failure rate is too high
    failureRate := float64(m.Failed) / float64(m.TotalEvents)
    if failureRate > 0.05 { // > 5% failures
        alerting.Send(fmt.Sprintf("High replay failure rate: %.1f%%", failureRate*100))
    }
}

// ✗ Replay without idempotency — produces double processing
for _, e := range dlqEvents {
    processPayment(e) // no check whether it was already processed!
}
// ✓ Always check event_id before reprocessing

// ✗ DLQ without metadata — hard to debug during investigation
dlq.Send(Message{Body: originalMsg.Body})
// ✓ Include the failure reason, attempt count, and full timestamp
dlq.Send(DLQMessage{
    Body:          originalMsg.Body,
    FailureReason: err.Error(),
    AttemptCount:  originalMsg.ReceiveCount,
    FailedAt:      time.Now(),
    StackTrace:    debug.Stack(),
})

// ✗ Mass replay without a kill switch — can't stop it in an emergency
func replayAllDLQ() {
    for _, e := range getAllEvents() {
        process(e) // no way to stop if the system starts struggling
    }
}
// ✓ Always use context as the kill switch
func replayAllDLQ(ctx context.Context) {
    for _, e := range getAllEvents() {
        if ctx.Err() != nil {
            log.Info("replay cancelled")
            return
        }
        process(ctx, e)
    }
}

// ✗ Events stored only in an in-memory queue — lost on restart
queue := make(chan Event, 1000) // not persistent
// ✓ A durable event store: PostgreSQL, S3, or Kafka with a long retention

// ✗ Side effects without a guard — email sent on every replay
func handleOrderConfirmed(event Event) {
    emailService.Send(event.UserEmail) // sent every time it's called
}
// ✓ Check state first, side effects only if the state hasn't changed