Activity and RPEI Flow

Last updated: 2026-07-25, JIM v0.14.0

This diagram shows how Activities are created, how Run Profile Execution Items (RPEIs) are accumulated during operations, and how the final activity status is determined. Activities are the immutable audit record for every operation in JIM.

Activities arise from two distinct paths. Operational work (synchronisation, data generation, clearing or deleting a Connected System, Temporal Scope Reconciliation) runs on a Worker Task, so its Activity is created when the task is queued and stays InProgress until the task completes. Configuration changes (issue #14) take no Worker Task: the config server that mutates the entity creates and completes the Activity synchronously, in the same call, alongside a versioned configuration snapshot.

Since v0.10.0, sync RPEIs are bulk-inserted per page via raw SQL (FlushRpeisAsync) rather than held in-memory for the full run, to keep memory bounded at 100K+ object scale. Summary statistics are accumulated incrementally during each flush. Cross-page reference resolution merges new attribute-flow rows under the existing MvoChange parent rather than creating a duplicate RPEI, resolving the previous ~2x RPEI duplication when references spanned multiple pages.

Activity Status Values

Status	Value	Meaning
NotSet	0	Default, should not appear in practice
InProgress	1	Set at creation, operation is running
Complete	2	All RPEIs succeeded, no errors
CompleteWithWarning	3	Some RPEIs have errors, but not all
CompleteWithError	4	Exception during processing (with stack trace)
FailedWithError	5	All RPEIs errored, or unhandled exception
Cancelled	6	User cancelled the operation

Activity Creation

flowchart TD
    Trigger{Task<br/>origin?}

    Trigger -->|Schedule fires| Scheduler[SchedulerServer:<br/>Queue step group tasks]
    Trigger -->|Manual run| Web[JIM.Web:<br/>User clicks Run]
    Trigger -->|API call| Api[API controller:<br/>Create task request]
    Trigger -->|Config change| ConfigServer[Config server e.g.<br/>ConnectedSystemServer, SchedulerServer,<br/>MetaverseServer, SearchServer, ServiceSettingsServer]

    Scheduler --> CreateTask[TaskingServer.CreateWorkerTaskAsync]
    Web --> CreateTask
    Api --> CreateTask

    CreateTask --> TaskType{Worker task<br/>type?}
    TaskType -->|SynchronisationWorkerTask| SyncActivity[TargetType = ConnectedSystemRunProfile<br/>TargetName = runProfile.Name<br/>TargetContext = connectedSystem.Name]
    TaskType -->|ExampleDataTemplateWorkerTask| DataGenActivity[TargetType = ExampleDataTemplate]
    TaskType -->|ClearConnectedSystemObjectsWorkerTask| ClearActivity[TargetType = ConnectedSystem<br/>TargetOperationType = Clear]
    TaskType -->|DeleteConnectedSystemWorkerTask| DeleteActivity[TargetType = ConnectedSystem<br/>TargetOperationType = Delete]
    TaskType -->|TemporalScopeReconciliationWorkerTask| TemporalActivity[TargetType = TemporalScopeReconciliation]

    SyncActivity --> CreateActivity
    DataGenActivity --> CreateActivity
    ClearActivity --> CreateActivity
    DeleteActivity --> CreateActivity
    TemporalActivity --> CreateActivity

    %% --- Configuration change: created synchronously, no worker task (#14) ---
    ConfigServer --> ConfigActivity[Create + complete Activity SYNCHRONOUSLY<br/>no worker task involved<br/>CreateActivityWithTriadAsync, then<br/>capture configuration snapshot, then<br/>CompleteActivityAsync in the same call]
    ConfigActivity --> ConfigTargets[TargetType is a configuration type:<br/>SynchronisationRule, Schedule, ServiceSetting,<br/>TrustedCertificate, ApiKey, Role, PredefinedSearch,<br/>ConnectorDefinition, MetaverseObjectType,<br/>MetaverseAttribute, ExampleDataSet, ObjectMatchingRule, ...<br/>TargetOperationType = Create, Update or Delete]
    ConfigTargets --> ConfigDone([Configuration change<br/>Activity persisted])

    CreateActivity[CreateActivityWithTriadAsync:<br/>Status = InProgress<br/>Executed = UtcNow<br/>Copy initiator triad from task<br/>Copy schedule execution context]
    CreateActivity --> Validate[ValidateActivity:<br/>InitiatedByType must not be NotSet<br/>User/ApiKey must have InitiatedById]
    Validate --> Persist[Persist Activity<br/>Associate with WorkerTask]

RPEI Accumulation During Import

flowchart TD
    ImportStart([Import processing]) --> SeparateList[RPEIs stored in SEPARATE list<br/>Not added to Activity.RunProfileExecutionItems<br/>Prevents EF Core from following<br/>Activity -> RPEI -> CSO navigation<br/>during SaveChanges]

    SeparateList --> PerObject{For each<br/>imported object}

    PerObject --> Success{Object<br/>outcome?}
    Success -->|New CSO| AddedRPEI[RPEI: ObjectChangeType = Added]
    Success -->|Updated CSO| UpdatedRPEI[RPEI: ObjectChangeType = Updated]
    Success -->|Obsoleted CSO| DeletedRPEI[RPEI: ObjectChangeType = Deleted]
    Success -->|No changes| NoRPEI[No RPEI created<br/>Avoids unnecessary allocations]

    PerObject --> Error{Error<br/>type?}
    Error -->|Duplicate attributes| DupAttrRPEI[RPEI: DuplicateImportedAttributes]
    Error -->|Unknown object type| TypeRPEI[RPEI: CouldNotMatchObjectType]
    Error -->|Duplicate external ID| DupObjRPEI[RPEI: DuplicateObject]
    Error -->|Missing external ID| MissingIdRPEI[RPEI: MissingExternalIdAttributeValue]
    Error -->|Unhandled exception| UnhandledRPEI[RPEI: UnhandledError<br/>+ stack trace]

    AddedRPEI --> AfterPersist
    UpdatedRPEI --> AfterPersist
    DeletedRPEI --> AfterPersist
    DupAttrRPEI --> AfterPersist
    TypeRPEI --> AfterPersist
    DupObjRPEI --> AfterPersist
    MissingIdRPEI --> AfterPersist
    UnhandledRPEI --> AfterPersist

    AfterPersist[After CSOs persisted:<br/>activity.AddRunProfileExecutionItems<br/>from separate list]
    AfterPersist --> UpdateActivity[UpdateActivityAsync<br/>RPEIs now safely attached]

RPEI Accumulation During Sync

flowchart TD
    SyncStart([Sync processing]) --> DirectAdd[RPEIs added per-page to<br/>activity.RunProfileExecutionItems<br/>during per-CSO processing,<br/>then bulk-inserted and cleared<br/>by FlushRpeisAsync]

    DirectAdd --> PerCSO{For each CSO}

    PerCSO --> TryCatch[Three-layer try-catch<br/>around ProcessConnectedSystemObjectAsync]

    TryCatch --> Normal{Normal<br/>outcome?}
    Normal -->|Projected| ProjectedRPEI[RPEI: ObjectChangeType = Projected]
    Normal -->|Joined| JoinedRPEI[RPEI: ObjectChangeType = Joined]
    Normal -->|Attribute Flow| FlowRPEI[RPEI: ObjectChangeType = AttributeFlow]
    Normal -->|No changes| SkipRPEI[No RPEI created<br/>Only when HasChanges = true]

    TryCatch --> JoinError{SyncJoin<br/>Exception?}
    JoinError -->|AmbiguousMatch| AmbiguousRPEI[RPEI: AmbiguousMatch error]
    JoinError -->|ExistingJoin| ExistingRPEI[RPEI: CouldNotJoinDueToExistingJoin]

    TryCatch --> Unhandled{Unhandled<br/>exception?}
    Unhandled -->|Yes| UnhandledRPEI2[RPEI: UnhandledError<br/>+ stack trace<br/>Processing continues to next CSO]

    PerCSO --> Obsolete{CSO<br/>obsolete?}
    Obsolete -->|Yes| ObsoleteRPEIs[Up to 2 RPEIs:<br/>Disconnected + Deleted]

RPEI Accumulation During Export

flowchart TD
    ExportStart([Export completes]) --> ProcessResults[ProcessExportResultAsync<br/>Creates RPEIs from<br/>ProcessedExportItems]

    ProcessResults --> PerExport{For each<br/>export result}

    PerExport --> Outcome{Export<br/>outcome?}
    Outcome -->|Create succeeded| ProvRPEI[RPEI: ObjectChangeType = Exported]
    Outcome -->|Update succeeded| ExpRPEI[RPEI: ObjectChangeType = Exported]
    Outcome -->|Delete succeeded| DeprovRPEI[RPEI: ObjectChangeType = Deprovisioned]
    Outcome -->|Failed| FailRPEI[RPEI: ErrorType = UnhandledError<br/>Error message + retry count]

Activity Status Determination

flowchart TD
    TaskDone([Task completes]) --> CalcStats[CalculateActivitySummaryStats:<br/>Count RPEIs by ObjectChangeType<br/>Populate TotalProjected, TotalJoined,<br/>TotalAttributeFlows, TotalErrors, etc.]

    CalcStats --> CheckErrors{Analyse<br/>RPEI errors}

    CheckErrors --> HasErrors{Any RPEI has<br/>ErrorType set<br/>and != NotSet?}
    HasErrors -->|No| CompleteOk[CompleteActivityAsync<br/>Status = Complete]

    HasErrors -->|Yes| AllErrors{ALL RPEIs<br/>have errors?}
    AllErrors -->|Yes| FailActivity[FailActivityWithErrorAsync<br/>Status = FailedWithError<br/>All items experienced an error]
    AllErrors -->|No, some| WarnActivity[CompleteActivityWithWarningAsync<br/>Status = CompleteWithWarning]

    CalcStats -.->|Exception thrown<br/>during status check| SafeFail[SafeFailActivityAsync<br/>See triple fallback below]

SafeFailActivityAsync - Triple Fallback

When activity completion fails (e.g., EF tracking corruption, disposed DbContext), this three-level fallback ensures activities are never left stuck in InProgress.

flowchart TD
    Error([Exception during<br/>activity completion]) --> Level1[Level 1: Normal<br/>FailActivityWithErrorAsync<br/>via ActivityServer]
    Level1 --> L1Result{Success?}
    L1Result -->|Yes| Done([Activity marked failed])

    L1Result -->|No| Level2[Level 2: Direct repository<br/>Update activity status directly<br/>Bypasses EF tracking issues]
    Level2 --> L2Result{Success?}
    L2Result -->|Yes| Done

    L2Result -->|No| Level3[Level 3: Emergency<br/>Create fresh JimApplication<br/>+ new DbContext<br/>Force-update activity status]
    Level3 --> L3Result{Success?}
    L3Result -->|Yes| Done
    L3Result -->|No| Fatal[FATAL: Log error<br/>Activity stuck in InProgress<br/>Requires manual intervention]

Key Design Decisions

• RPEI list separation during import
  RPEIs are maintained in a separate list during import to prevent EF Core from following the Activity -> RPEI -> CSO navigation chain during SaveChanges. This avoids accidentally persisting CSOs before they're ready.

• Direct attachment during sync
  During sync, RPEIs are added directly to activity.RunProfileExecutionItems since CSOs already exist in the database (they were created during a prior import).

• Conditional RPEI creation
  RPEIs are only created when HasChanges = true during sync. This is an optimisation to avoid unnecessary allocations for objects that haven't changed.

• Error isolation
  Each CSO is processed within its own try-catch during sync. Errors create RPEIs but do not halt processing of remaining CSOs. This ensures a single bad object doesn't prevent the entire sync from completing.

• Three-tier status model
  Complete (no errors), CompleteWithWarning (some errors), FailedWithError (all errors or unhandled exception). This gives operators clear visibility into the severity of issues.

• Triple fallback for failure
  SafeFailActivityAsync ensures activities are never left stuck in InProgress, even when the DbContext is corrupted or disposed. This is critical for system reliability; stuck activities would block future schedule executions.

• Initiator triad audit
  Every activity records who initiated it (InitiatedByType, InitiatedById, InitiatedByName). For scheduled tasks, this preserves the schedule context. For deferred MVO deletions, the original initiator is captured at mark time and replayed during housekeeping.

• Synchronous configuration-change Activities (#14)
  Configuration changes do not go through a Worker Task. The owning config server creates the Activity, mutates the entity, captures a configuration snapshot, and completes the Activity all in the same synchronous call. ActivityTargetType now spans the full configuration surface, including SynchronisationRule, Schedule, ServiceSetting, TrustedCertificate, ApiKey, Role, PredefinedSearch, ConnectorDefinition, MetaverseObjectType, MetaverseAttribute, ObjectMatchingRule and ExampleDataSet, in addition to the operational target types created by Worker Tasks.

• Incremental stat counters (#1078)
  GetActivityRunProfileExecutionStatsAsync no longer aggregates over the RPEI and outcome tables on every read. The persistence paths maintain advisory per-Activity counter rows as the run executes, so an in-progress Activity's stats (served repeatedly while an administrator watches a run) are an O(counter rows) lookup. On completion, FinaliseActivityRunProfileExecutionStatsAsync replaces the counters with an exact aggregation and sets Activity.RunProfileExecutionStatsFinalised. Activities that completed before the counter table existed keep the legacy aggregation path and are finalised lazily the first time their stats are read. Non-relational providers (the EF in-memory test provider) always use the aggregation path, since counter maintenance is raw SQL.

• Scope-exit and no-contributor sync outcomes (Attribute Priority, #91)
  Beyond the ObjectChangeType recorded on each RPEI, sync builds a finer-grained outcome tree. A CSO leaving scope (rather than being deleted at source) records a DisconnectedOutOfScope outcome, and when a recalled attribute has no surviving contributor and is genuinely cleared (not re-elected to a survivor, not frozen under a deletion grace period), a NoContributor child outcome is surfaced so an administrator can see the blank was an event, not merely an uncontributed attribute.