Convergence Policies

The SDK can drive desired→reported convergence for you. You declare a policy per writable property path; the SDK pushes through the adapter, observes reported, retries with exponential backoff, classifies upstream errors, and records the outcome in desiredSync.

This is the answer to "what if I want a property to keep trying until reported reports back the same value?".

When to use

Reach for a convergence policy whenever pushing a desired property to the device requires external coordination and is not guaranteed to succeed on the first try:

• REST aggregators (Enode, Tuya, Shelly Cloud, …) — upstream may rate-limit, return 5xx, or reject with 4xx ("already charging") that the cloud cannot see in advance.
• Slow physical devices — set-points that take seconds or minutes to reach the desired value (thermostats, dimmers, valves).
• Async upstream actions — the upstream returns 202 + an action id and reaches the desired state later.

Skip convergence policies for properties that converge atomically on the first push (a simple MQTT retain message, a fire-and-forget config blob).

Anatomy of a policy

A policy declares five things:

1. Path — a dotted JSON path under desired ("chargingMode", "config.interval").
2. HasConverged(desired, reported) — predicate over the desired and reported subtrees rooted at Path. Returns true once the device matches.
3. Backoff — exponential schedule with optional jitter, capped at a max delay.
4. Timeout — hard upper bound; the scheduler stamps ac=504, ad="Timeout" and stops if exceeded.
5. Outcome routing — what to do when the adapter throws:
   • OnRejected (default Fail) — applies to UpstreamRejectedException (4xx-class).
   • OnError (default Retry) — applies to any other exception (5xx, network, timeout).

Registering a policy

using BeaconTower.ProviderSdk.Convergence;

builder.Services.AddBeaconTowerProvider<MyAdapter>(builder, options =>
{
    options.ProviderId = "my-provider";
});

builder.Services.AddConvergencePolicy(b => b
    .ForPath("chargingMode")
        .ConvergesWhen((desired, reported) =>
            desired is JsonValue dv
            && reported is JsonValue rv
            && dv.GetValue<string>() == rv.GetValue<string>())
        .Backoff(new BackoffSchedule(
            initial: TimeSpan.FromSeconds(10),
            max:     TimeSpan.FromMinutes(2),
            factor:  2.0,
            jitter:  0.2))
        .Timeout(TimeSpan.FromMinutes(5))
        .OnRejected(ConvergenceOutcome.Fail) // 4xx → stop, ac=4xx
        .OnError(ConvergenceOutcome.Retry)   // 5xx → keep trying
        .Done());

You can chain multiple paths in a single block:

.AddConvergencePolicy(b => b
    .ForPath("chargingMode").ConvergesWhen(...).Done()
    .ForPath("config.interval").ConvergesWhen(...).Done());

Lifecycle

PATCH /api/devices/{id}/desired { "chargingMode": "START" }
  │
  ▼
TwinManagementService applies merge-patch, increments desiredVersion
  │
  ▼
DesiredDeliveryTracker pushes via adapter (existing behavior)
  │
  ▼  on success
ConvergenceScheduler.OnDesiredDelivered(deviceId, version)
  │
  ├─► reported already matches  ──► desiredSync.X = { ac: 200, av: N, ad: "Converged" }
  │
  └─► not converged             ──► desiredSync.X = { ac: 202, av: N, ad: "Pending", attempts: 0 }
                                      │
                                      ▼  after backoff(1)
                                    re-fetch twin, re-evaluate HasConverged
                                      ├─► converged   → ac: 200, stop
                                      ├─► timeout     → ac: 504, stop
                                      ├─► path gone   → clear envelope, stop
                                      ├─► superseded  → another run owns the slot, stop
                                      └─► not yet     → re-push via adapter
                                                          ├─► UpstreamRejectedException(4xx)
                                                          │     └─► OnRejected
                                                          │           ├─► Fail  → ac: 4xx, stop
                                                          │           └─► Stop  → clear, stop
                                                          ├─► other Exception
                                                          │     └─► OnError (default Retry)
                                                          └─► success → schedule next backoff

Adapter responsibilities

To play nicely with convergence, the adapter's SendDesiredPropertiesAsync:

1. Push the desired value upstream. Same as before.
2. Throw UpstreamRejectedException(statusCode, reason) for 4xx-class definitive rejections — the convergence scheduler stamps the envelope with that exact status and reason, then stops retrying (per OnRejected).
3. Throw any other exception for transient failures (5xx, network, timeout). The scheduler retries per OnError (default Retry) within the configured timeout.

public override async Task SendDesiredPropertiesAsync(
    string deviceId, JsonObject properties, CancellationToken ct)
{
    using var resp = await _http.PostAsJsonAsync(uri, body, ct);

    if ((int)resp.StatusCode is >= 400 and < 500)
    {
        var reason = await resp.Content.ReadAsStringAsync(ct);
        throw new UpstreamRejectedException((int)resp.StatusCode, reason);
    }

    resp.EnsureSuccessStatusCode(); // 5xx → HttpRequestException → retry
}

Reading the result

Bus-api callers and frontends watch desiredSync to know whether a PATCH stuck. Typical UI logic:

ac	Meaning	UI hint
(no envelope)	Not under convergence control	Plain reported value
200	Converged	Green checkmark
202	Pending — keep watching	Spinner / "in progress"
4xx	Upstream rejected	Red "Action rejected"
504	Timed out	Yellow "Did not confirm"

Always check av against the current desiredVersion — an envelope with av < desiredVersion is stale; the scheduler will reseed shortly.

Restart durability

The scheduler rehydrates pending envelopes on host start: any twin with desiredSync.X.ac == 202 and av >= desiredVersion is re-armed. Stale envelopes (av < desiredVersion) are skipped — the next desired write reseeds them.

Backoff schedule

new BackoffSchedule(
    initial: TimeSpan.FromSeconds(5),   // delay before first retry
    max:     TimeSpan.FromMinutes(2),   // cap on any single delay
    factor:  2.0,                       // each retry × factor
    jitter:  0.2);                      // ±20% randomization

Retry n (1-indexed) waits roughly initial × factor^(n-1), capped at max, multiplied by a random factor in [1-jitter, 1+jitter].

BackoffSchedule.Default is a sensible starting point: 5s → 2min, factor 2, 20% jitter.

Observability

Metrics on the BeaconTower.ProviderSdk.Convergence meter:

Metric	Type	Tags
convergence.attempts.total	Counter	path
convergence.outcome.total	Counter	outcome ∈ converged, rejected, timeout, error_terminal, superseded, stopped
convergence.duration_ms	Histogram	outcome

The meter is registered automatically when you use the BeaconTower service defaults.

Example — Enode

The reference BeaconTower.EnodeProvider example uses this pattern for chargingMode:

builder.Services.AddConvergencePolicy(b => b
    .ForPath("chargingMode")
        .ConvergesWhen((desired, reported) =>
            desired is JsonValue dv
            && reported is JsonValue rv
            && dv.GetValue<string>() == rv.GetValue<string>())
        .Backoff(new BackoffSchedule(
            TimeSpan.FromSeconds(10),
            TimeSpan.FromMinutes(2),
            factor: 2.0,
            jitter: 0.2))
        .Timeout(TimeSpan.FromMinutes(5))
        .OnRejected(ConvergenceOutcome.Fail)
        .OnError(ConvergenceOutcome.Retry)
        .Done());

The Enode HTTP client throws UpstreamRejectedException for 4xx (e.g. "already charging" → 409 → desiredSync.chargingMode.ac = 409, ad: "...") and lets HttpRequestException bubble for 5xx so the SDK retries.

Limitations

• The current scheduler keeps in-flight state in memory plus the persisted desiredSync envelope. Distributed (multi-instance) coordination is out of scope — the broadcast subscriber pattern from twin-management already ensures a single instance handles delivery; convergence runs on that instance only.
• Rehydration on startup is a single-page scan of twins with non-null desiredSync. Large fleets (tens of thousands of in-flight envelopes) will benefit from a partial JSONB index — file an issue if this becomes a bottleneck.
• Cross-property dependencies (e.g. "wait for mode to converge before retrying power") are not modelled; each policy is independent.

See also

• desiredSync — the shadow hierarchy
• Adapter interface
• Error handling