Sync Lifecycle & Error Handling

This document covers the complete sync lifecycle for API-based integrations, including triggers, job flow, state management, and error recovery.

Sync Trigger Methods

1. Manual Sync (User-initiated)

Advisor clicks “Sync” button in UI.

Single Household:

POST /advisors/{advisor}/integrations/{integration}/mappings/{mapping}/sync

Entire Integration (limited vendors):

POST /advisors/{advisor}/integrations/{integration}/sync

Only supported for: Advyzon, Morningstar Office, Riskalyze, Redtail, Wealthbox

2. Nightly Sync (Scheduled)

Background job runs during off-peak hours to sync all active integrations.

Scenario: Scenario::NIGHTLY_SYNC

Flow:

1. Scheduler triggers nightly sync command
2. Query all integrations where failed_biz_days < threshold
3. Dispatch sync job for each integration
4. Rate limit per vendor to avoid API throttling

3. Webhook Trigger (Vendor Push)

External vendor or internal service pushes update notification.

Location: app/Http/Controllers/Webhooks/Integrations/

Supported Vendors:

• File-based vendors via AWS SNS (TriggerSyncController) - triggered when Collector finishes processing files
• AssetBook - direct webhook
• LPL - direct webhook

Note: Yodlee does NOT support webhook triggers. It uses a passive refresh model where data only syncs when users actively trigger a refresh via FastLink or manual refresh button.

Flow:

1. Receive webhook POST (or SNS notification)
2. Validate signature/auth
3. Parse payload, identify affected integration
4. Dispatch targeted sync job

Integration Scenarios

Defined in app/Integrations/Support/Scenario.php as PHP attribute for tracking:

enum Scenario: string
{
    case CREATE_INTEGRATION = 'create_integration';   // Initial setup
    case LINK_ENTITY        = 'link_entity';          // Link vendor entity to household
    case LIST_ENTITIES      = 'list_entities';        // Discover available accounts
    case SYNC_HOUSEHOLD     = 'sync_household';       // Sync single household
    case SYNC_INTEGRATION   = 'sync_integration';     // Sync all accounts
    case NIGHTLY_SYNC       = 'nightly_sync';         // Scheduled background sync
    case IMPORT_HOUSEHOLD   = 'import_household';     // Bulk household import
    case IMPORT_CLIENTS     = 'import_clients';       // Import client data
    case LIST_TAGS          = 'list_tags';            // Get CRM tags
    case SEND_PDF           = 'send_pdf';             // Document delivery
}

Usage in Controllers:

class IntegrationMappingSyncController extends Controller
{
    #[IntegrationScenario(Scenario::SYNC_HOUSEHOLD)]
    public function store(Request $request, Advisor $advisor, Integration $integration, IntegrationMapping $mapping): Response
    {
        // Sync logic here
    }
}

Job Queue Flow

┌─────────────────────────────────────────────────────────────────────────────┐
│                           Trigger (Manual/Nightly/Webhook)                   │
└─────────────────────────────────────────────────────────────────────────────┘
                                    │
                                    ▼
┌─────────────────────────────────────────────────────────────────────────────┐
│                          Controller / Command                                │
│                                                                              │
│   1. Validate request                                                        │
│   2. Load Integration + IntegrationMapping                                   │
│   3. Check if sync allowed (rate limit, not already syncing)                │
│   4. Dispatch Job (async) or run inline (sync)                              │
└─────────────────────────────────────────────────────────────────────────────┘
                                    │
                                    ▼
┌─────────────────────────────────────────────────────────────────────────────┐
│                              Queue Worker                                    │
│                                                                              │
│   1. Pick job from queue                                                     │
│   2. Set integration.is_syncing = true                                       │
│   3. Allocate memory (some integrations need 2GB+)                          │
└─────────────────────────────────────────────────────────────────────────────┘
                                    │
                                    ▼
┌─────────────────────────────────────────────────────────────────────────────┐
│                           Integrator::sync()                                 │
│                                                                              │
│   1. Initialize Connector                                                    │
│   2. Authenticate (OAuth token refresh if needed)                           │
│   3. Fetch data from vendor API                                             │
│   4. Parse response into vendor models                                       │
│   5. Map to internal models (Account, Insurance, etc.)                      │
│   6. Create/Update/Delete internal records                                   │
│   7. Update IntegrationMapping.last_completed_at                            │
└─────────────────────────────────────────────────────────────────────────────┘
                                    │
                    ┌───────────────┴───────────────┐
                    ▼                               ▼
            ┌───────────┐                   ┌───────────┐
            │  Success  │                   │  Failure  │
            └───────────┘                   └───────────┘
                    │                               │
                    ▼                               ▼
┌─────────────────────────────┐   ┌─────────────────────────────────────────┐
│ integration.failed_since    │   │ integration.failed_since = now()        │
│           = null            │   │ integration.failed_biz_days++           │
│ integration.last_completed  │   │ integration.failure_message = error     │
│           = now()           │   │                                         │
│ integration.is_syncing      │   │ Log error to Sentry                     │
│           = false           │   │                                         │
└─────────────────────────────┘   └─────────────────────────────────────────┘

State Management

Integration State Fields

$integration->is_syncing;        // bool - Currently syncing (computed from queue)
$integration->failed_since;      // Carbon|null - When failures started
$integration->failed_biz_days;   // int (0-255) - Consecutive business days failed
$integration->failure_message;   // string|null - User-friendly error message
$integration->last_completed_at; // Carbon|null - Last successful sync

IntegrationMapping State Fields

$mapping->sync_status;           // string - Current sync status
$mapping->last_completed_at;     // Carbon|null - Last successful sync for this mapping
$mapping->deleted_at;            // Carbon|null - Soft delete timestamp
$mapping->deleted_by_user_id;    // int|null - Who deleted it (audit trail)

Sync Status Flow

                    ┌──────────┐
                    │  IDLE    │
                    └────┬─────┘
                         │ sync triggered
                         ▼
                    ┌──────────┐
                    │ SYNCING  │
                    └────┬─────┘
                         │
            ┌────────────┼────────────┐
            ▼            ▼            ▼
       ┌────────┐   ┌────────┐   ┌────────┐
       │SUCCESS │   │ FAILED │   │PARTIAL │
       └────┬───┘   └────┬───┘   └────┬───┘
            │            │            │
            └────────────┼────────────┘
                         ▼
                    ┌──────────┐
                    │  IDLE    │ (ready for next sync)
                    └──────────┘

Error Handling & Recovery

Exception Hierarchy

Exception (abstract base)
├── ExternalServiceException (vendor API errors)
│   ├── ConnectException          // Connection failed
│   ├── TimeoutException          // Request timeout
│   ├── UnauthorizedException     // Auth failed (401)
│   ├── ForbiddenException        // Access denied (403)
│   ├── NotFoundException         // Entity not found (404)
│   ├── TooManyRequestException   // Rate limited (429)
│   ├── ServiceUnavailableException // Vendor down (503)
│   └── UnrecognizedException     // Unknown error

Error Recovery Strategies

Exception	AffectedLevel	Action
UnauthorizedException	Integration	Mark integration failed, prompt user to reconnect
ForbiddenException	Integration	Mark integration failed, check permissions
NotFoundException	Mapping	Soft delete mapping, account removed at vendor
TooManyRequestException	Integration	Retry with exponential backoff
TimeoutException	Mapping	Retry up to 3 times
ServiceUnavailableException	Integration	Retry later, vendor is down

Failure Tracking Mechanism

// On sync failure
if ($exception instanceof ExternalServiceException) {
    if ($integration->failed_since === null) {
        $integration->failed_since = now();
    }

    // Increment only on business days
    if (today()->isWeekday()) {
        $integration->failed_biz_days++;
    }

    $integration->failure_message = $exception->getDefaultHttpResponseMessage();
}

// Auto-disable threshold (e.g., 5 business days)
if ($integration->failed_biz_days >= 5) {
    // Stop nightly syncs, require user intervention
    $integration->failure_message = 'Integration disabled after repeated failures. Please reconnect.';
}

Success Reset

// On successful sync
$integration->failed_since = null;
$integration->failed_biz_days = 0;
$integration->failure_message = null;
$integration->last_completed_at = now();

$mapping->last_completed_at = now();
$mapping->sync_status = 'success';

Retry Logic

Automatic Retries

class SyncIntegrationJob implements ShouldQueue
{
    public int $tries = 3;
    public int $backoff = 60; // seconds

    public array $backoffStrategy = [60, 300, 900]; // 1min, 5min, 15min

    public function handle(): void
    {
        try {
            $this->integration->getIntegratorFqcn()::sync($this->mapping);
        } catch (TooManyRequestException $e) {
            // Release back to queue with delay
            $this->release($e->getRetryAfter() ?? 60);
        } catch (TimeoutException $e) {
            // Retry immediately (will use backoff)
            throw $e;
        }
    }

    public function failed(Throwable $exception): void
    {
        // All retries exhausted
        $this->integration->update([
            'failed_since' => $this->integration->failed_since ?? now(),
            'failure_message' => $exception->getMessage(),
        ]);

        // Report to Sentry
        if ($exception->shouldReportToSentry()) {
            report($exception);
        }
    }
}

Rate Limiting

// Prevent too many syncs per integration
$rateLimitKey = $integration->getRateLimitedKeyForNightlySync();

if (RateLimiter::tooManyAttempts($rateLimitKey, 1)) {
    // Already syncing or recently synced
    return;
}

RateLimiter::hit($rateLimitKey, 3600); // 1 hour cooldown

Memory Management

Some integrations handle large datasets and need increased memory:

#[IntegrationScenario(Scenario::LINK_ENTITY)]
public function store(Request $request, Advisor $advisor, Integration $integration): Response
{
    // Allocate 2GB for large datasets
    ini_set('memory_limit', '2048M');

    // ... process large account list
}

Debugging & Logging

Integration Event Logging

// Integrator base class provides file logging
protected static function logToFile(?string $reference, string $postfix, array $data): void
{
    $path = storage_path("logs/integrations/{$reference}_{$postfix}.json");
    file_put_contents($path, json_encode($data, JSON_PRETTY_PRINT));
}

// Usage in sync
public static function sync(IntegrationMapping $mapping): ?array
{
    $response = $connector->call('/accounts');

    // Log raw API response for debugging
    static::logToFile(
        $mapping->reference,
        'api_response_' . now()->format('Y-m-d_H-i-s'),
        $response
    );

    // ... process response
}

Sentry Integration

Exceptions that should be reported implement shouldReportToSentry():

abstract class Exception extends \Exception
{
    public function shouldReportToSentry(): bool
    {
        // Most vendor errors should be reported
        return true;
    }
}

// Some exceptions are expected and shouldn't flood Sentry
class NotFoundException extends ExternalServiceException
{
    public function shouldReportToSentry(): bool
    {
        return false; // Account deleted at vendor is expected
    }
}

Related Documentation

• Patterns - Connector and Integrator design
• Vendor Specifics - Yodlee state machine, Schwab pagination
• Architecture Overview - System overview