BeepSync Manager

Overview

BeepSyncManager is the modern replacement for the legacy DataSyncManager. It is a partial class split across nine focused files and orchestrates one-way or bidirectional data synchronization between any two registered IDataSource instances.

Folder map

Editor/BeepSync/
  BeepSyncManager.Core.cs              constructor, properties, dispose
  BeepSyncManager.Sync.cs              main SyncDataAsync() pipeline
  BeepSyncManager.BulkSync.cs          SyncAllData*, SyncData overloads
  BeepSyncManager.Cdc.cs               BuildCdcFilterContext()
  BeepSyncManager.ConflictResolution.cs TryEvaluateConflictGate()
  BeepSyncManager.Preflight.cs         RunPreflightAsync()
  BeepSyncManager.SchemaManagement.cs  Add/Remove/Update, Save/Load
  BeepSyncManager.RetryCheckpoint.cs   TryClassifyError, IsCheckpointResumeSafe
  BeepSyncManager.SchemaGovernance.cs  PromoteSchemaAsync()
  SyncMetrics.cs                       metrics/telemetry data model
  SyncRuleExecutionPolicies.cs         DefaultSafe / FastPath profiles
  Interfaces/
    ISyncHelpers.cs                    helper interfaces
  Helpers/
    DataSourceHelper.cs                IDataSource open/get/insert/update
    FieldMappingHelper.cs              AutoMapFields, type conversion
    SyncValidationHelper.cs            schema & watermark validation
    SyncProgressHelper.cs              progress reporting & run history
    SchemaPersistenceHelper.cs         JSON file save/load/backup
  Models/
    CdcFilterContext.cs                CDC filter result envelope
    SyncIntegrationContext.cs          optional rule/defaults/mapping wiring
    SyncPlanMetadata.cs                run-time plan metadata
  Examples/
    Example01_BasicFullSync.cs
    Example02_SchemaVersioning.cs
    Example03_IncrementalSync.cs
    Example04_BidirectionalConflict.cs
    Example05_RetryAndCheckpoint.cs
    Example06_DataQualityAndReconciliation.cs
    Example07_SloAndAlerting.cs
    Example08_PerformanceAndParallel.cs

Key design principles

Migration from DataSyncManager

Quick Start

Minimal example — one-way full sync of a Customers table.

// 1. Create the manager
var syncManager = new BeepSyncManager(dmeEditor);

// 2. Describe the sync
var schema = new DataSyncSchema
{
    Id                        = Guid.NewGuid().ToString(),
    SourceDataSourceName      = "SourceDB",
    DestinationDataSourceName = "DestDB",
    SourceEntityName          = "Customers",
    DestinationEntityName     = "Customers",
    SourceKeyField            = "CustomerId",
    DestinationKeyField       = "CustomerId",
    SyncType                  = "Full",       // "Full" | "Incremental"
    SyncDirection             = "OneWay",     // "OneWay" | "Bidirectional"
    BatchSize                 = 500,
};

// 3. Auto-map fields (name matching)
var fieldHelper = new FieldMappingHelper(dmeEditor);
foreach (var m in fieldHelper.AutoMapFields("SourceDB", "Customers", "DestDB", "Customers"))
    schema.MappedFields.Add(m);

// 4. Register & run
syncManager.AddSyncSchema(schema);
var progress = new Progress<PassedArgs>(p => Console.WriteLine(p.Messege));
await syncManager.SyncDataAsync(schema, CancellationToken.None, progress);

// 5. Inspect result
Console.WriteLine($"Status: {schema.SyncStatus}  LastSync: {schema.LastSyncDate:u}");

// 6. Persist
await syncManager.SaveSchemasAsync();

Core API — Constructor & Properties

Constructors

// Minimal — helpers wired internally
BeepSyncManager(IDMEEditor editor)

// With optional rule/defaults integration context
BeepSyncManager(IDMEEditor editor, SyncIntegrationContext integrationContext)

SyncIntegrationContext

When set, unlocks rule-engine gating for every policy decision point.

var ctx = new SyncIntegrationContext
{
    RuleEngine      = myRuleEngine,     // IRuleEngine — drives all rule policy gates
    DefaultsManager = myDefaults,       // IDefaultsManager — fills default field values
    RunInitiatedBy  = "scheduler",      // stamped on schema versions & audit logs
    // MappingPlanState = ...           // optional: shared mapping plan state
};
var syncManager = new BeepSyncManager(editor, ctx);

Properties

Schema Management

CRUD methods

// Add a new schema to the in-memory list
syncManager.AddSyncSchema(schema);

// Update an existing schema by Id
syncManager.UpdateSyncSchema(schema);

// Remove by Id
syncManager.RemoveSyncSchema(schemaId);

// Manual schema validation (returns IErrorsInfo)
IErrorsInfo result = syncManager.ValidateSchema(schema);

Filters & field mappings

// Add a filter clause (AppFilter wraps a field/operator/value predicate)
syncManager.AddFilter(schemaId, new AppFilter
{
    FieldName  = "Region",
    Operator   = "=",
    FilterValue = "EU"
});

// Remove a filter by field name
syncManager.RemoveFilter(schemaId, "Region");

// Add a single field mapping
syncManager.AddFieldMapping(schemaId, new FieldSyncData
{
    SourceField      = "CustomerName",
    DestinationField = "Name",
});

Persistence

// Async — preferred
await syncManager.SaveSchemasAsync();
await syncManager.LoadSchemasAsync();

// Sync fallback
syncManager.SaveSchemas();
syncManager.LoadSchemas();

DataSyncSchema — key fields

SyncDataAsync — Full Orchestration

Signature

Task SyncDataAsync(
    DataSyncSchema         schema,
    CancellationToken      token          = default,
    IProgress<PassedArgs> progress       = null,
    IImportErrorStore      errorStore     = null,
    IImportHistoryStore    historyStore   = null
);

Pipeline steps

When called, SyncDataAsync executes the following ordered pipeline:

Bulk Sync

When you have multiple schemas registered, the bulk helpers iterate them in one call.

Sequential

// Async — processes schemas one after another
await syncManager.SyncAllDataAsync(CancellationToken.None, progress);

// Sync fallback
syncManager.SyncAllData();

Parallel

// Runs schemas in parallel, bounded by SyncPerformanceProfile.MaxDegreeOfParallelism
await syncManager.SyncAllDataParallelAsync(CancellationToken.None, progress);

The semaphore count is read from the schema's performance profile rule result (default 4). This avoids overwhelming shared data sources under load.

SyncData overloads (sync, by schema ID)

Preflight Validation

Method

Task<SyncPreflightReport> RunPreflightAsync(
    DataSyncSchema    schema,
    CancellationToken token = default
);

Checks performed

SyncPreflightReport

var report = await syncManager.RunPreflightAsync(schema);

if (!report.IsApproved)
{
    foreach (var issue in report.Issues)
        Console.WriteLine($"[{issue.Severity}] {issue.Code}: {issue.Message}");
    return; // block the run
}

Console.WriteLine($"Rules passed : {report.RulesPassed}");
Console.WriteLine($"Defaults ready: {report.DefaultsReady}");

CDC & Watermark (Incremental Sync)

Set SyncType = "Incremental" and configure WatermarkPolicy to restrict each sync run to only records changed since the previous run.

WatermarkPolicy

schema.WatermarkPolicy = new WatermarkPolicy
{
    WatermarkMode        = "Timestamp",   // "Timestamp" | "Sequence" | "CompositeKey"
    WatermarkField       = "UpdatedAt",   // column that advances monotonically
    LastWatermarkValue   = null,          // null = start from beginning; engine updates this
    OverlapWindowSeconds = 300,           // re-scan last 5 min to catch late arrivals
    DedupeStrategy       = "LastWrite",   // "LastWrite" | "FirstWrite" | "None"
    ReplayEnabled        = true,          // idempotent: safe to re-run the same window

    // Optional: rule keys override the built-in filter/late-arrival logic
    FilterRuleKey        = "sync.watermark.filter",
    LateArrivalRuleKey   = "sync.watermark.late-arrival",
};

WatermarkMode values

CdcFilterContext (output)

Returned by BuildCdcFilterContext — the runtime filter envelope passed to DataImportManager.

Conflict Resolution

Applies only when SyncDirection = "Bidirectional". When a record exists in both source and destination with different values, TryEvaluateConflictGate decides which value wins.

ConflictPolicy

schema.ConflictPolicy = new ConflictPolicy
{
    // Built-in rule keys (no rule engine required):
    //   "sync.conflict.source-wins"
    //   "sync.conflict.destination-wins"
    //   "sync.conflict.latest-timestamp-wins"
    //   "sync.conflict.fail-on-conflict"
    ResolutionRuleKey     = "sync.conflict.latest-timestamp-wins",

    CaptureEvidence       = true,    // populate BeepSyncManager.LastRunConflicts

    // Quarantine records whose conflict could not be resolved:
    QuarantineDsName      = "QuarantineDB",
    QuarantineEntity      = "ConflictQuarantine",

    MaxConflictsPerRun    = 500,
    OnMaxExceededAction   = "QuarantineRest",  // "Abort" | "Continue" | "QuarantineRest"
};

Built-in strategies

ConflictEvidence (audit)

// After a bidirectional run:
var conflicts = syncManager.LastRunConflicts;
Console.WriteLine($"Conflicts this run: {conflicts.Count}");

foreach (var ev in conflicts)
{
    Console.WriteLine($"  Key={ev.RecordKey}  " +
                      $"Resolution={ev.Resolution}  " +
                      $"Rule={ev.AppliedRuleKey}");
}

Retry & Checkpoint

Configure RetryPolicy to automatically re-attempt on transient errors and resume from a mid-run checkpoint without re-processing already-committed batches.

RetryPolicy

schema.RetryPolicy = new RetryPolicy
{
    MaxAttempts              = 4,
    BaseDelayMs              = 2000,
    BackoffMode              = "Exponential",  // "Fixed" | "Linear" | "Exponential"
    MaxResumeWindowHours     = 48,             // checkpoint invalidated after 48 h

    CheckpointEnabled        = true,           // save offset after each committed batch

    // Rule Engine key that classifies errors into categories
    ErrorCategoryRuleKey     = "sync.error.classify",

    NonRetryableCategories   = new List<string> { "Fatal", "Validation" },
};

Error categories

If the rule key ErrorCategoryRuleKey is not set or the rule engine is disabled, TryClassifyError applies keyword heuristics (e.g. "timeout" → Transient, "constraint" → Validation).

Checkpoint resume

// After a partial failure the checkpoint is stored in schema.ActiveCheckpoint.
// On the next call to SyncDataAsync, the manager calls IsCheckpointResumeSafe()
// which evaluates the rule "sync.checkpoint.resume-safe" or checks the age window.

// Force a full re-run by clearing the checkpoint:
schema.ActiveCheckpoint = null;
await syncManager.SyncDataAsync(schema, CancellationToken.None, progress);

Schema Governance & Promotion

PromoteSchemaAsync advances a schema through an approval lifecycle (e.g. Draft → Review → Approved). A rule engine gate can block promotion if the mapping quality or approval evidence is insufficient.

PromoteSchemaAsync

Task<SyncPreflightReport> PromoteSchemaAsync(
    DataSyncSchema    schema,
    string            targetState,    // e.g. "Review", "Approved", "Archived"
    string            approver   = null,
    string            changeNotes = null,
    CancellationToken token      = default
);

Promotion pipeline

Example

var report = await syncManager.PromoteSchemaAsync(
    schema,
    targetState  : "Approved",
    approver     : "data-steward@example.com",
    changeNotes  : "Reviewed mapping completeness — 100%"
);

if (report.IsApproved)
    Console.WriteLine("Schema promoted to Approved.");
else
    foreach (var issue in report.Issues)
        Console.WriteLine($"  BLOCKED: {issue.Code} — {issue.Message}");

Sync Metrics

SyncMetrics is the telemetry model emitted (and stored via SyncProgressHelper) after every run.

SyncMetrics properties

Rule Execution Policies

SyncRuleExecutionPolicies provides two pre-built RuleExecutionPolicy profiles used internally when invoking the rule engine at each gate.

// The schema's performance profile mode selects the policy at runtime:
var policy = SyncRuleExecutionPolicies.Resolve(schema.PerformanceProfile?.RulePolicyMode);
// "FastPath" → FastPath profile; anything else → DefaultSafe

Built-in rule keys used by BeepSyncManager

Helper Classes

All five helpers are instantiated by BeepSyncManager internally. You can also instantiate them directly for unit testing or standalone use.

AutoMapFields

var fieldHelper = new FieldMappingHelper(editor);

// Returns List<FieldSyncData> for all columns with matching names
var mappings = fieldHelper.AutoMapFields(
    "SourceDB", "Customers",
    "DestDB",   "Customers");

// Override specific mappings after auto-map:
var nameMapping = mappings.First(m => m.SourceField == "ContactName");
nameMapping.DestinationField = "FullName";

ValidateWatermarkPolicy

var validationHelper = new SyncValidationHelper(editor);

var result = validationHelper.ValidateWatermarkPolicy(schema);
if (result.Flag == Errors.Failed)
    Console.WriteLine($"[ERROR] {result.Message}");

Examples

Eight runnable examples live in Editor/BeepSync/Examples/. Below is a representative selection.

Example 01 — Basic full sync

// Example01_BasicFullSync.cs
var syncManager = new BeepSyncManager(editor);
var schema = new DataSyncSchema
{
    Id                        = Guid.NewGuid().ToString(),
    SourceDataSourceName      = "SourceDB",
    DestinationDataSourceName = "DestDB",
    SourceEntityName          = "Customers",
    DestinationEntityName     = "Customers",
    SourceKeyField            = "CustomerId",
    DestinationKeyField       = "CustomerId",
    SyncType                  = "Full",
    SyncDirection             = "OneWay",
    BatchSize                 = 500,
};

var fieldHelper = new FieldMappingHelper(editor);
foreach (var m in fieldHelper.AutoMapFields("SourceDB","Customers","DestDB","Customers"))
    schema.MappedFields.Add(m);

syncManager.AddSyncSchema(schema);
await syncManager.SyncDataAsync(schema, CancellationToken.None,
    new Progress<PassedArgs>(p => Console.WriteLine(p.Messege)));

Console.WriteLine($"Status: {schema.SyncStatus}  LastSync: {schema.LastSyncDate:u}");

Example 03 — Incremental sync with watermark

// Example03_IncrementalSync.cs
var schema = new DataSyncSchema
{
    // ... source/dest/entity/key fields ...
    SyncType  = "Incremental",
    WatermarkPolicy = new WatermarkPolicy
    {
        WatermarkMode        = "Timestamp",
        WatermarkField       = "UpdatedAt",
        OverlapWindowSeconds = 300,
        DedupeStrategy       = "LastWrite",
        ReplayEnabled        = true,
    },
};

// First run — syncs records changed since epoch:
await syncManager.SyncDataAsync(schema, CancellationToken.None, null);

// Second run — only picks up records changed since the first run:
await syncManager.SyncDataAsync(schema, CancellationToken.None, null);

Example 04 — Bidirectional conflict resolution

// Example04_BidirectionalConflict.cs
schema.SyncDirection  = "Bidirectional";
schema.WatermarkPolicy = new WatermarkPolicy
    { WatermarkMode = "Timestamp", WatermarkField = "ModifiedAt", ReplayEnabled = true };

schema.ConflictPolicy = new ConflictPolicy
{
    ResolutionRuleKey   = "sync.conflict.latest-timestamp-wins",
    CaptureEvidence     = true,
    QuarantineDsName    = "QuarantineDB",
    QuarantineEntity    = "ConflictQuarantine",
    MaxConflictsPerRun  = 500,
    OnMaxExceededAction = "QuarantineRest",
};

await syncManager.SyncDataAsync(schema, CancellationToken.None, null);

foreach (var ev in syncManager.LastRunConflicts)
    Console.WriteLine($"Key={ev.RecordKey}  Resolution={ev.Resolution}");

Example 05 — Retry with checkpoint resume

// Example05_RetryAndCheckpoint.cs
schema.RetryPolicy = new RetryPolicy
{
    MaxAttempts           = 4,
    BaseDelayMs           = 2000,
    BackoffMode           = "Exponential",
    MaxResumeWindowHours  = 48,
    CheckpointEnabled     = true,
    ErrorCategoryRuleKey  = "sync.error.classify",
    NonRetryableCategories = new List<string> { "Fatal", "Validation" },
};

// Run 1 — saves a checkpoint if it fails mid-way
await syncManager.SyncDataAsync(schema, CancellationToken.None, progress);

// Run 2 — resumes from the checkpoint automatically if it is still within 48 h
await syncManager.SyncDataAsync(schema, CancellationToken.None, progress);

// Force full re-run:
schema.ActiveCheckpoint = null;
await syncManager.SyncDataAsync(schema, CancellationToken.None, progress);

Example 02 — Schema versioning & promotion

// Example02_SchemaVersioning.cs
var ctx = new SyncIntegrationContext
{
    RuleEngine     = myRuleEngine,
    RunInitiatedBy = "data-steward@example.com",
};
var syncManager = new BeepSyncManager(editor, ctx);

// ... add schema, configure mappings ...

var report = await syncManager.PromoteSchemaAsync(
    schema,
    targetState  : "Approved",
    approver     : "data-steward@example.com",
    changeNotes  : "All mappings reviewed and complete."
);

if (report.IsApproved)
    Console.WriteLine("Schema is now Approved and versioned.");
else
    foreach (var issue in report.Issues)
        Console.WriteLine($"[BLOCKED] {issue.Code}: {issue.Message}");

All examples summary

Full API Reference

BeepSyncManager — Complete Public Surface

Core Properties

SyncMetrics — Complete Properties (23)

ISyncValidationHelper — Complete Interface

ISyncProgressHelper — Complete Interface

ISchemaPersistenceHelper — Complete Interface

*Additional members beyond ISchemaPersistenceHelper interface.

IDataSourceHelper & IFieldMappingHelper

public interface IDataSourceHelper
{
    IDataSource GetDataSource(string name);
    bool ValidateDataSourceConnection(string name);
    Task<object> GetEntityDataAsync(string ds, string entity, List<AppFilter>);
    Task<IErrorsInfo> InsertEntityAsync(string ds, string entity, object entity);
    Task<IErrorsInfo> UpdateEntityAsync(string ds, string entity, object entity);
    Task<bool> EntityExistsAsync(string ds, string entity, List<AppFilter>);
}

public interface IFieldMappingHelper
{
    List<FieldSyncData> AutoMapFields(string srcDS, string srcEntity, string destDS, string destEntity);
    IErrorsInfo ValidateFieldMappings(IEnumerable<FieldSyncData>);
    IEnumerable<FieldSyncData> LoadGovernedMapping(DataSyncSchema);
    bool CheckMappingDrift(DataSyncSchema, IEnumerable<FieldSyncData> baseline);
    IErrorsInfo PromoteMappingState(DataSyncSchema, string targetState);
}

SyncSchemaTranslator — Static API

public static class SyncSchemaTranslator
{
    // Converts BeepSync schema → DataImportManager configuration
    public static DataImportConfiguration ToImportConfiguration(
        DataSyncSchema schema,
        IImportErrorStore errorStore = null,
        IImportRunHistoryStore historyStore = null);

    // Converts BeepSync schema → reverse-direction DataImportManager configuration
    public static DataImportConfiguration ToReverseImportConfiguration(
        DataSyncSchema schema,
        IImportErrorStore errorStore = null,
        IImportRunHistoryStore historyStore = null);
}

SyncRuleExecutionPolicies

public static class SyncRuleExecutionPolicies
{
    public static RuleExecutionPolicy DefaultSafe;  // MaxDepth=10, MaxExecutionMs=5000
    public static RuleExecutionPolicy FastPath;      // MaxDepth=3, MaxExecutionMs=2000

    public static RuleExecutionPolicy Resolve(string mode);
    // Returns FastPath if mode == "FastPath" (case-insensitive), otherwise DefaultSafe
}

Model Types

CdcFilterContext

SyncIntegrationContext

SyncPlanMetadata

Source File Map (29 files)