Recovery: Live Sync¶

Discord Gateway 에서 발생한 길드 상태 변경 이벤트를 실시간 수집하고, 5초 주기 배치로 DB 에 반영한다. 복구의 기반이 되는 "현재 길드 상태" 의 source of truth 를 유지한다.

Sub-context position¶

Parent — Recovery Core domain
Location — engine/recovery/subdomain/sync/
Sibling sub-contexts — Snapshot, Restore, AntiNuke

Why this sub-context exists¶

Umbra 의 스냅샷/복구가 제대로 동작하려면 "우리 DB 에 저장된 길드 상태" 가 실제 Discord 길드와 거의 일치해야 한다. Sync 는 이 일치를 유지하는 책임을 진다.

순진한 구현(이벤트당 즉시 UPSERT)은 결제 SaaS 규모에서 문제가 있다. Discord 에서는 역할 재배열, 대량 멤버 추가/제거 같은 변경이 짧은 시간에 폭주 한다. 매 이벤트마다 DB 쓰기를 하면 connection pool 고갈과 중복 쓰기가 발생한다.

Sync 는 배치 + merge 로 이 문제를 해결한다 (ADR-0023). 이벤트를 Redis 에 버퍼링하고, 5초 주기로 길드별 배치를 처리하며, 같은 aggregate 에 대한 연속 변경은 최종 상태만 DB 에 반영한다.

Domain model¶

DiscordEvent (value object)¶

Bot 프로세스가 수신한 Discord Gateway 이벤트.

DiscordEvent
├─ event_type        TEXT        'GUILD_MEMBER_UPDATE' 등
├─ timestamp         TIMESTAMPTZ
├─ guild_id          TEXT        Discord snowflake
├─ aggregate_type    TEXT        'member' | 'role' | 'channel' | 'permission_override' | 'guild'
├─ aggregate_id      TEXT        Discord snowflake (member id, role id, etc.)
└─ payload           JSONB       이벤트별 변경 데이터

GuildState (aggregate)¶

길드의 현재 상태. Recovery domain 이 소유 (source of truth). 4개 관련 테이블:

GuildRole            roles 상태
├─ guild_id, discord_role_id (PK)
├─ name, color, permissions, position, hoist, mentionable
└─ updated_at

GuildChannel         channels 상태
├─ guild_id, discord_channel_id (PK)
├─ name, type, parent_id, position, topic, nsfw, rate_limit
└─ updated_at

GuildPermissionOverride  permission overrides
├─ guild_id, channel_id, target_type, target_id (PK)
├─ allow_bits, deny_bits
└─ updated_at

GuildSettings        guild-level 설정
├─ guild_id (PK)
├─ name, icon_hash, afk_channel_id, afk_timeout, verification_level, …
└─ updated_at

SyncBuffer (value object)¶

Redis 에 저장된 길드별 이벤트 버퍼. Key: sync:guild:{guild_id}:buffer.

SyncCheckpoint¶

(optional) 마지막 처리 시각 기록. 장애 복구 후 buffer 가 어느 시점부터인지 추정하기 위함.

Invariants¶

Guild 당 단일 poller — 같은 길드의 버퍼를 두 워커가 동시 처리 못 하도록 Redis distributed lock (sync:guild:{guild_id}:lock, TTL 30s)
Buffer size cap — 1000 events 초과 시 즉시 flush
Pause 중 이벤트는 버퍼에 유지 — 드롭 안 함
Resume 시 버퍼 선택적 clear — Restore 직후는 clear (구 상태 반영 방지)

State machine¶

SyncBuffer 는 길드 단위로 상태를 가진다.

stateDiagram-v2
    [*] --> Active : Bot installed
    Active --> Paused : Restore started signal
    Paused --> Active : Restore completed signal (buffer cleared)
    Active --> Stopped : Bot kicked / license inactive
    Paused --> Stopped : Bot kicked during restore
    Stopped --> Active : Bot reinstalled + license active

Domain events¶

Published¶

Sync 는 도메인 이벤트를 직접 발행하지 않는다. Snapshot/Restore/AntiNuke 가 간접 영향을 받을 뿐이다. 단 감사 목적으로 다음은 기록:

Event	Trigger	Payload
`LiveSyncBatchProcessed` (internal)	배치 처리 완료	`guild_id, events_count, merged_count, duration_ms`

내부용이라 Outbox 가 아닌 metric 으로 수집.

Consumed¶

Source	Channel	Action
Discord Gateway events	Bot 프로세스 내부 함수 호출	버퍼에 push
`Restore.Pause` signal (Temporal)	Temporal signal	버퍼 pause
`Restore.Resume` signal (Temporal)	Temporal signal	버퍼 clear + resume

Event flow¶

sequenceDiagram
    participant Bot as umbra-bot
    participant Redis as Redis buffer
    participant Cron as asynq cron (5s)
    participant Worker as sync worker
    participant DB as Neon PostgreSQL

    Bot->>Redis: LPUSH sync:guild:G:buffer {event}
    Note over Redis: 버퍼에 이벤트 누적

    Cron->>Worker: FlushGuildBuffer(G)
    Worker->>Redis: SET NX sync:guild:G:lock (TTL 30s)
    alt Lock acquired
        Worker->>Redis: LRANGE buffer (up to 1000)
        Redis-->>Worker: events
        Worker->>Worker: Merge by aggregate_id
        Worker->>DB: BEGIN
        Worker->>DB: UPSERT roles, channels, overrides, settings
        Worker->>DB: COMMIT
        Worker->>Redis: LTRIM buffer (처리한 만큼 제거)
        Worker->>Redis: DEL sync:guild:G:lock
    else Lock busy
        Note over Worker: skip this cycle
    end

Merge logic¶

Aggregate 단위 최종 상태 계산¶

같은 (aggregate_type, aggregate_id) 에 대한 이벤트는 순서대로 적용 후 최종 상태만 저장.

events:
  ROLE_UPDATE role=X { color: red }
  ROLE_UPDATE role=X { name: "Admin" }
  ROLE_DELETE role=X

merged:
  ROLE_DELETE role=X  (DELETE 는 모든 이전 변경을 무효화)

events:
  MEMBER_UPDATE member=M { roles: [A, B] }
  MEMBER_UPDATE member=M { roles: [A, B, C] }
  MEMBER_UPDATE member=M { nickname: "Bob" }

merged:
  UPSERT member=M { roles: [A, B, C], nickname: "Bob" }

규칙:

DELETE 이벤트가 있으면 DELETE 로 수렴 (이전 UPDATE 무시)
UPDATE 는 필드별 last-write-wins (병합)
CREATE + UPDATE + DELETE 가 한 배치에 있으면 최종적으로 DELETE (row 생성 생략)

Batch 처리 예시¶

type mergeResult struct {
    upserts map[string]AggregateState  // aggregate_id → final state
    deletes map[string]struct{}
}

func merge(events []DiscordEvent) mergeResult {
    // 순서대로 accumulate
    // DELETE 는 upserts 에서 key 제거, deletes 에 추가
    ...
}

Ports¶

Inbound¶

// engine/recovery/subdomain/sync/port/service.go
type SyncService interface {
    EnqueueEvent(ctx, guildID uuid.UUID, event DiscordEvent) error
    FlushBuffer(ctx, guildID uuid.UUID) (*FlushResult, error)
    PauseBuffer(ctx, guildID uuid.UUID) error
    ResumeBuffer(ctx, guildID uuid.UUID, clearBuffer bool) error
    IsPaused(ctx, guildID uuid.UUID) (bool, error)
}

type FlushResult struct {
    EventsProcessed int
    AggregatesMerged int
    Duration time.Duration
}

Outbound¶

type EventBuffer interface {
    Push(ctx, guildID, event) error
    PopBatch(ctx, guildID, max) ([]DiscordEvent, error)
    Length(ctx, guildID) (int, error)
    TryLock(ctx, guildID, ttl) (bool, error)
    Unlock(ctx, guildID) error
    MarkPaused(ctx, guildID, paused bool) error
    IsPaused(ctx, guildID) (bool, error)
    Clear(ctx, guildID) error
}

type GuildStateRepository interface {
    UpsertRole(ctx, guildID, role RoleState) error
    DeleteRole(ctx, guildID, discordRoleID) error
    UpsertChannel(ctx, guildID, channel ChannelState) error
    DeleteChannel(ctx, guildID, discordChannelID) error
    UpsertPermissionOverride(ctx, guildID, override OverrideState) error
    DeletePermissionOverride(ctx, guildID, channelID, targetType, targetID) error
    UpsertGuildSettings(ctx, guildID, settings GuildSettingsState) error

    // 트랜잭션 경계
    WithTx(ctx, fn func(txRepo GuildStateRepository) error) error
}

Adapters¶

Persistence — engine/recovery/subdomain/sync/adapter/persistence/ — sqlc 기반, 트랜잭션 래퍼 포함
Redis buffer — engine/recovery/subdomain/sync/adapter/redis/ — LIST + distributed lock
Event ingestion — apps/bot/internal/handler/gateway.go 에서 수신 후 EnqueueEvent 호출

Member sync¶

Member 는 Recovery 가 아닌 Member 도메인 소유다. 단 Sync 는 Member 상태도 관찰해야 하므로 다음 전략:

Discord member 이벤트는 Sync 가 수신
Member 도메인의 UpsertMember / MarkMemberLeft 를 호출
즉 Sync 는 Recovery domain 상태 + Member domain 상태 둘 다 기록

이를 위해 Sync 는 MemberWriter 포트를 추가 의존:

type MemberWriter interface {
    UpsertMember(ctx, guildID, discordUserID, roles, nickname) error
    MarkMemberLeft(ctx, guildID, discordUserID) error
}

Member 수정도 같은 트랜잭션에서 커밋 (cross-schema 트랜잭션).

Permission model¶

Live Sync 는 RECOVERY_LIVE_SYNC feature 로 게이팅.

Free: 비활성 — Bot 은 이벤트 수신하지만 DB 에 저장 안 함 (메모리 drop)
Pro/Enterprise: 활성

이는 Bot 프로세스의 gateway 핸들러가 licensing.Can(guildID, FeatureRecoveryLiveSync) 를 체크 후 EnqueueEvent 호출 여부 결정.

Pause / Resume flow¶

Restore workflow 가 시작될 때 Sync 는 반드시 pause 되어야 한다.

sequenceDiagram
    participant Restore as Restore Workflow
    participant Sync as Sync Service
    participant Redis

    Restore->>Sync: PauseBuffer(guildID)
    Sync->>Redis: SET sync:guild:G:paused = 1
    Sync-->>Restore: OK

    Note over Sync: 이후 배치 워커는 paused 확인 후 skip
    Note over Restore: Restore 단계 진행 (수십 초 ~ 수 분)

    Restore->>Sync: ResumeBuffer(guildID, clearBuffer=true)
    Sync->>Redis: DEL sync:guild:G:buffer (복구 전 이벤트 제거)
    Sync->>Redis: DEL sync:guild:G:paused
    Sync-->>Restore: OK

clearBuffer=true 인 이유: Restore 중 쌓인 이벤트는 "복구 이전 상태" 를 반영하므로, 적용하면 방금 복구한 최신 상태를 덮어씌우게 된다. 버퍼를 비우고 resume 후에 새 이벤트만 받는다.

Failure modes¶

Redis 장애 — 버퍼 불가, 이벤트는 Bot 메모리에서 drop. 복구 후에도 과거 이벤트 손실. Discord 재연결 후 INIT sync 트리거 로 전체 상태 재동기화 (Phase 2 기능, MVP 는 손실 감수).
Worker 프로세스 장애 — 다른 워커가 lock 획득 후 처리
Lock holder crash (lock TTL 전) — TTL 30s 후 자동 해제
DB 트랜잭션 실패 — LTRIM 을 하지 않아 이벤트 유지, 다음 cycle 에 재시도
이벤트 폭주 (burst) — 1000 초과 시 즉시 flush, 이후 cron 대기 없이 연속 flush
Paused 상태에서 Bot kicked — Sync 는 Stopped. 복구 완료 시점에 Guild 도 없으므로 이후 작업 무의미.