282 lines
10 KiB
Go
282 lines
10 KiB
Go
package inbound
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"log"
|
|
"strings"
|
|
"time"
|
|
)
|
|
|
|
// Recovery only replays API-owned pending deliveries; Redis recovery locks and
|
|
// presence snapshots prevent multiple Gateway instances from racing the replay.
|
|
|
|
type pendingDeliveryRequest struct {
|
|
Account string `json:"account"`
|
|
Limit int `json:"limit,omitempty"`
|
|
}
|
|
|
|
type pendingDelivery struct {
|
|
ID string `json:"id"`
|
|
DeliveryType string `json:"deliveryType"`
|
|
Payload json.RawMessage `json:"payload"`
|
|
CreatedAt time.Time `json:"createdAt"`
|
|
}
|
|
|
|
type pendingFlushResult struct {
|
|
Account string
|
|
Deliveries int
|
|
DeliveredCount int
|
|
FailedCount int
|
|
WaitingCount int
|
|
LastError string
|
|
}
|
|
|
|
func (s Server) flushPending(account string, logger *log.Logger) (pendingFlushResult, error) {
|
|
result := pendingFlushResult{Account: account}
|
|
if account == "" {
|
|
return result, nil
|
|
}
|
|
var deliveries []pendingDelivery
|
|
if err := s.post(context.Background(), "/gateway/events/downstream/pending", pendingDeliveryRequest{Account: account, Limit: 100}, &deliveries); err != nil {
|
|
logger.Printf("cmpp inbound pending delivery fetch failed account=%s err=%v", account, err)
|
|
result.LastError = err.Error()
|
|
return result, err
|
|
}
|
|
result.Deliveries = len(deliveries)
|
|
for _, delivery := range deliveries {
|
|
sendResult, err := s.pushPendingDelivery(account, delivery)
|
|
if err != nil {
|
|
result.FailedCount++
|
|
result.LastError = err.Error()
|
|
_ = s.post(context.Background(), "/gateway/events/downstream/failed", map[string]any{
|
|
"id": delivery.ID, "errorMessage": err.Error(), "failureType": "send_failed",
|
|
"connectionId": sendResult.ConnectionID, "sequenceId": sendResult.SequenceID,
|
|
"messageId": sendResult.MessageID, "sentAt": sendResult.SentAt,
|
|
}, nil)
|
|
continue
|
|
}
|
|
if sendResult.Sent {
|
|
result.DeliveredCount++
|
|
continue
|
|
}
|
|
if sendResult.ReasonCode == "SUBMIT_RESPONSE_PENDING" {
|
|
result.WaitingCount++
|
|
continue
|
|
}
|
|
errorMessage := defaultString(sendResult.ErrorMessage, "gateway did not complete downstream delivery")
|
|
failureType := "unrecoverable"
|
|
if sendResult.Retryable {
|
|
failureType = "send_failed"
|
|
result.WaitingCount++
|
|
} else {
|
|
result.FailedCount++
|
|
}
|
|
result.LastError = errorMessage
|
|
_ = s.post(context.Background(), "/gateway/events/downstream/failed", map[string]any{
|
|
"id": delivery.ID, "errorMessage": errorMessageWithCode(errorMessage, sendResult.ReasonCode),
|
|
"failureType": failureType, "connectionId": sendResult.ConnectionID,
|
|
"sequenceId": sendResult.SequenceID, "messageId": sendResult.MessageID, "sentAt": sendResult.SentAt,
|
|
}, nil)
|
|
}
|
|
return result, nil
|
|
}
|
|
|
|
func (s Server) pushPendingDelivery(account string, delivery pendingDelivery) (DownstreamSendResult, error) {
|
|
switch delivery.DeliveryType {
|
|
case "receipt":
|
|
var event DownstreamReceipt
|
|
if err := json.Unmarshal(delivery.Payload, &event); err != nil {
|
|
return DownstreamSendResult{}, err
|
|
}
|
|
event.DeliveryID = delivery.ID
|
|
event.Account = defaultString(event.Account, account)
|
|
allowRecovery := !delivery.CreatedAt.IsZero() && time.Since(delivery.CreatedAt) >= 5*time.Second
|
|
return pushReceiptWithResult(event, allowRecovery)
|
|
case "uplink":
|
|
var event DownstreamUplink
|
|
if err := json.Unmarshal(delivery.Payload, &event); err != nil {
|
|
return DownstreamSendResult{}, err
|
|
}
|
|
event.DeliveryID = delivery.ID
|
|
event.Account = defaultString(event.Account, account)
|
|
return PushUplinkWithResult(event)
|
|
default:
|
|
return DownstreamSendResult{}, fmt.Errorf("unsupported downstream delivery type %q", delivery.DeliveryType)
|
|
}
|
|
}
|
|
|
|
func (s Server) runPendingFlusher(logger *log.Logger) {
|
|
ticker := time.NewTicker(s.pendingFlushInterval())
|
|
defer ticker.Stop()
|
|
for range ticker.C {
|
|
s.flushOnlineAccounts(logger)
|
|
s.recoverPendingCandidates(logger)
|
|
}
|
|
}
|
|
|
|
func (s Server) flushOnlineAccounts(logger *log.Logger) {
|
|
for _, account := range onlineAccounts() {
|
|
_, _ = s.flushPending(account, logger)
|
|
}
|
|
}
|
|
|
|
func (s Server) recoverPendingCandidates(logger *log.Logger) {
|
|
candidates, err := ListRecoveryCandidates(context.Background(), s.PresenceStore)
|
|
if err != nil {
|
|
logger.Printf("cmpp inbound recovery candidate refresh failed err=%v", err)
|
|
return
|
|
}
|
|
for _, candidate := range candidates {
|
|
account := strings.TrimSpace(candidate.Account)
|
|
if account == "" {
|
|
continue
|
|
}
|
|
activeRecovery := DownstreamRecoveryStatus{
|
|
Account: account,
|
|
GatewayInstanceID: s.gatewayInstanceID(),
|
|
}
|
|
if s.RecoveryStore != nil {
|
|
decision, recoveryErr := s.RecoveryStore.StartAccountRecovery(context.Background(), account, s.gatewayInstanceID())
|
|
if recoveryErr != nil {
|
|
logger.Printf("cmpp inbound recovery start failed account=%s err=%v", account, recoveryErr)
|
|
continue
|
|
}
|
|
if !decision.Allowed {
|
|
logger.Printf("cmpp inbound recovery skipped account=%s reason=%s", account, decision.SkipReason)
|
|
decision.Status.Account = account
|
|
decision.Status.GatewayInstanceID = s.gatewayInstanceID()
|
|
decision.Status.State = defaultString(decision.Status.State, "failed")
|
|
decision.Status.LastSkipReason = defaultString(decision.Status.LastSkipReason, decision.SkipReason)
|
|
if decision.Status.FailureCategory == "" {
|
|
decision.Status.FailureCategory = recoveryFailureCategory(decision.Status.State, "", decision.Status.LastSkipReason)
|
|
}
|
|
s.syncRecoveryStatus(logger, account, decision.Status)
|
|
continue
|
|
}
|
|
activeRecovery = decision.Status
|
|
activeRecovery.Account = account
|
|
activeRecovery.GatewayInstanceID = s.gatewayInstanceID()
|
|
}
|
|
result, flushErr := s.flushPending(account, logger)
|
|
if s.RecoveryStore != nil {
|
|
status := DownstreamRecoveryStatus{
|
|
Account: account,
|
|
GatewayInstanceID: s.gatewayInstanceID(),
|
|
LockToken: activeRecovery.LockToken,
|
|
LockOwner: defaultString(activeRecovery.LockOwner, s.gatewayInstanceID()),
|
|
LockAcquiredAt: activeRecovery.LockAcquiredAt,
|
|
LockExpiresAt: activeRecovery.LockExpiresAt,
|
|
LastAttemptAt: activeRecovery.LastAttemptAt,
|
|
}
|
|
switch {
|
|
case flushErr != nil:
|
|
status.State = "failed"
|
|
status.LastError = flushErr.Error()
|
|
status.FailureCategory = recoveryFailureCategory(status.State, status.LastError, status.LastSkipReason)
|
|
case result.WaitingCount > 0 && result.DeliveredCount == 0 && result.FailedCount == 0:
|
|
status.State = "waiting_connection"
|
|
status.LastError = "downstream client is not connected"
|
|
status.FailureCategory = recoveryFailureCategory(status.State, status.LastError, status.LastSkipReason)
|
|
case result.FailedCount > 0 && result.DeliveredCount > 0:
|
|
status.State = "partial"
|
|
status.LastError = result.LastError
|
|
status.FailureCategory = recoveryFailureCategory(status.State, status.LastError, status.LastSkipReason)
|
|
case result.FailedCount > 0:
|
|
status.State = "failed"
|
|
status.LastError = result.LastError
|
|
status.FailureCategory = recoveryFailureCategory(status.State, status.LastError, status.LastSkipReason)
|
|
default:
|
|
status.State = "success"
|
|
status.LastError = ""
|
|
status.FailureCategory = ""
|
|
}
|
|
if err := s.RecoveryStore.CompleteAccountRecovery(context.Background(), status); err != nil {
|
|
logger.Printf("cmpp inbound recovery completion failed account=%s err=%v", account, err)
|
|
if err == ErrRecoveryLockLost {
|
|
status.State = "failed"
|
|
status.LastSkipReason = "lock_lost"
|
|
status.FailureCategory = "lock_lost"
|
|
s.syncRecoveryStatus(logger, account, status)
|
|
}
|
|
} else if persisted, err := s.RecoveryStore.GetAccountRecoveryStatus(context.Background(), account); err != nil {
|
|
logger.Printf("cmpp inbound recovery status fetch failed account=%s err=%v", account, err)
|
|
} else {
|
|
s.syncRecoveryStatus(logger, account, persisted)
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
func (s Server) syncRecoveryStatus(logger *log.Logger, account string, status DownstreamRecoveryStatus) {
|
|
if err := s.post(context.Background(), "/gateway/events/downstream/recovery-status", map[string]any{
|
|
"account": status.Account,
|
|
"gatewayInstanceId": status.GatewayInstanceID,
|
|
"state": status.State,
|
|
"lockOwner": status.LockOwner,
|
|
"lockExpiresAt": formatRFC3339Nano(status.LockExpiresAt),
|
|
"lastAttemptAt": formatRFC3339Nano(status.LastAttemptAt),
|
|
"lastSuccessAt": formatRFC3339Nano(status.LastSuccessAt),
|
|
"lastFailureAt": formatRFC3339Nano(status.LastFailureAt),
|
|
"nextRetryAt": formatRFC3339Nano(status.NextRetryAt),
|
|
"attemptCount": status.AttemptCount,
|
|
"failureCategory": status.FailureCategory,
|
|
"lastError": status.LastError,
|
|
"lastSkipReason": status.LastSkipReason,
|
|
}, nil); err != nil {
|
|
logger.Printf("cmpp inbound recovery status sync failed account=%s err=%v", account, err)
|
|
}
|
|
}
|
|
|
|
func recoveryFailureCategory(state string, lastError string, lastSkipReason string) string {
|
|
if state == "success" || state == "running" {
|
|
return ""
|
|
}
|
|
if lastSkipReason == "backoff" {
|
|
return "backoff"
|
|
}
|
|
if lastSkipReason == "locked" {
|
|
return "lock_contended"
|
|
}
|
|
if lastSkipReason == "lock_lost" {
|
|
return "lock_lost"
|
|
}
|
|
if state == "waiting_connection" {
|
|
return "client_disconnected"
|
|
}
|
|
if state == "partial" {
|
|
return "partial_delivery_failed"
|
|
}
|
|
if state == "failed" && strings.TrimSpace(lastError) != "" {
|
|
return "flush_failed"
|
|
}
|
|
return "unknown"
|
|
}
|
|
|
|
func (s Server) pendingFlushInterval() time.Duration {
|
|
if s.PendingFlushInterval > 0 {
|
|
return s.PendingFlushInterval
|
|
}
|
|
return time.Minute
|
|
}
|
|
|
|
func (s Server) logRecoveryCandidates(logger *log.Logger) {
|
|
candidates, err := ListRecoveryCandidates(context.Background(), s.PresenceStore)
|
|
if err != nil {
|
|
logger.Printf("cmpp inbound recovery candidates load failed err=%v", err)
|
|
return
|
|
}
|
|
if len(candidates) == 0 {
|
|
logger.Printf("cmpp inbound recovery candidates loaded count=0")
|
|
return
|
|
}
|
|
accounts := make([]string, 0, len(candidates))
|
|
for _, item := range candidates {
|
|
if strings.TrimSpace(item.Account) != "" {
|
|
accounts = append(accounts, item.Account)
|
|
}
|
|
}
|
|
logger.Printf("cmpp inbound recovery candidates loaded count=%d accounts=%s", len(candidates), strings.Join(accounts, ","))
|
|
}
|