Connector Protocol Reference

Overview#

The Atlas connector wire is a single SignalR hub at /hubs/connector, serialized with MessagePack (LZ4-framed), carrying seven method names in both directions. This article is the canonical reference for those message shapes: every record, every [Key(n)] number, every [Union(n)] discriminator, and every enum ordinal.

Everything here is load-bearing. MessagePack writes these records positionally by key index and never by name, so a key number that disagrees with Atlas does not fail loudly - it decodes into the wrong field or throws deep inside the serializer with no reference to your type.

The protocol has two halves:

• Pull (backfill). Atlas invokes your registered GetRange handler and awaits a ConnectorRangeResponse. This is the only pull method.
• Push (live). Atlas invokes your registered Subscribe handler, you ack per item, and from then on your connector pushes unprompted with PushSubscriptionRows, PushSubscriptionStatus and PushSubscriptionWatermark.

Related reading: Connector Overview and Lifecycle covers the narrative lifecycle (connect, handshake, reconcile, subscribe, disconnect, reconnect). Connector Range Queries covers how Atlas segments and re-issues GetRange and how to translate a PushdownRequest into a query against your upstream. This article owns the shapes those two articles refer to.

There is no point-in-time pull#

Atlas never asks a connector for a single current value. There is no pull handler to implement.

Values travel in both directions as a structure-of-arrays rowset: the Column union inside a CompoundRowSet. Atlas serves a point-in-time scalar read from its own store, using the last row a live subscription pushed. So if you want a current value to appear in Atlas, implement Subscribe.

Mandatory client setup#

Three things must be right before any shape in this article matters. Each of them fails in a way that does not point at itself.

1. Bind ConnectorMessagePackOptions.Options on both ends#

A bare .AddMessagePackProtocol() connects successfully and then fails on every frame.

ConnectorMessagePackOptions.Options is:

public static readonly MessagePackSerializerOptions Options =
    MessagePackSerializerOptions.Standard
        .WithResolver(Resolver)                                  // :32-35
        .WithCompression(MessagePackCompression.Lz4BlockArray);

where Resolver is CompositeResolver.Create(NativeGuidResolver.Instance, StandardResolver.Instance).

Two independent hazards, both recorded in that file's own remarks:

NativeGuidResolver writes a Guid as a 16-byte bin payload instead of the default 36-character ASCII string. Atlas binds these options; the shipped client binds them. The two sides must be upgraded in lockstep.

2. Use the right port#

Atlas binds 4551 (HTTP) and 4552 (HTTPS), and the shipped connector configuration agrees. Atlas has never listened on 5214 or 8080.

3. Attach a bearer - the hub is gated by default#

:

app.MapHub<ConnectorHub>("/hubs/connector").RequireAtlasHubAuth(app.Configuration);

RequireAtlasHubAuth applies the AuthPolicies.HubAccess policy only when Pulse:Auth:Enabled is true - and Pulse:Auth:Enabled: true is the shipped default. HubAccess pins both the SPA cookie scheme and the JWT bearer scheme, so an unauthenticated negotiate gets a 401, not a 302 redirect.

A machine connector has no browser session, so it must present a Pulse client-credentials bearer through SignalR's HttpConnectionOptions.AccessTokenProvider. The connector's Pulse client registration must list Atlas's Pulse:Auth:ClientId (default "Atlas") as an audience, or the token will validate as a token but fail the HubAccess policy.

The shipped client assigns AccessTokenProvider only when Pulse:Auth:Enabled=true: SignalR invokes the delegate on every negotiate and every reconnect regardless of whether the hub challenges, so leaving it wired against an auth-disabled Atlas issues a pointless client_credentials POST on every connect and couples reconnect latency to Pulse.

The connect sample#

Compilable against net10.0 with Microsoft.AspNetCore.SignalR.Client 10.0.3 and Microsoft.AspNetCore.SignalR.Protocols.MessagePack 10.0.3 (the versions the shipped connector host uses,).

using Microsoft.AspNetCore.SignalR.Client;
using Microsoft.Extensions.DependencyInjection;   // AddMessagePackProtocol
using Timebase.Atlas.Shared.Ingestion;

// Supply your own Pulse client-credentials token here. Return null when
// Atlas runs with Pulse:Auth:Enabled=false.
static Task<string?> GetAtlasBearerAsync() => Task.FromResult<string?>(null);

var connection = new HubConnectionBuilder()
    .WithUrl("http://localhost:4551/hubs/connector", options =>
    {
        // /hubs/connector is gated by the HubAccess policy whenever
        // Pulse:Auth:Enabled=true, which is the shipped default.
        options.AccessTokenProvider = GetAtlasBearerAsync;
    })
    .AddMessagePackProtocol(opts =>
    {
        // MANDATORY. Omitting this connects, then fails on every frame
        // and mis-decodes every Guid.
        opts.SerializerOptions = ConnectorMessagePackOptions.Options;
    })
    // SignalR's default policy gives up after four attempts (0s, 2s, 10s, 30s),
    // which kills the connector permanently on any Atlas restart.
    .WithAutomaticReconnect(new IndefiniteRetryPolicy(TimeSpan.FromSeconds(5)))
    .Build();

// A wide CompoundRowSet on a long-polling transport can exceed the default
// 30-second idle timeout while the payload is still transferring.
connection.ServerTimeout = TimeSpan.FromMinutes(5);

// Register EVERY handler before StartAsync. Atlas can invoke Subscribe the
// moment the handshake returns.
connection.On<ConnectorRangeRequest, ConnectorRangeResponse>(
    ConnectorMethods.GetRange,
    request => HandleGetRange(request));

connection.On<IReadOnlyList<SubscriptionRequest>, IReadOnlyList<SubscriptionAck>>(
    ConnectorMethods.Subscribe,
    requests => HandleSubscribe(requests));

connection.On<IReadOnlyList<Guid>>(
    ConnectorMethods.Unsubscribe,
    ids => HandleUnsubscribe(ids));

await connection.StartAsync();

var advertisement = new ConnectorAdvertisement(
    ConnectorId: "conn.example",
    DisplayName: "Example Connector",
    Datasources: new[]
    {
        new DatasourceDescriptor(
            DatasourceId: "example-ds",
            Parameters:   Array.Empty<ParameterDescriptor>(),
            BindingPaths: new[] { "example/plant/pump1/speed" },
            Templates:    Array.Empty<string>()),
    });

var handshake = await connection.InvokeAsync<HandshakeResponse>(
    ConnectorMethods.Handshake, advertisement);

Console.WriteLine($"Atlas session {handshake.AtlasSessionId}");

// The client must re-handshake itself after every reconnect: SignalR's
// automatic reconnect does not replay hub invocations, and Atlas's
// OnDisconnectedAsync unregistered every binding path.
connection.Reconnected += async _ =>
{
    await connection.InvokeAsync<HandshakeResponse>(
        ConnectorMethods.Handshake, advertisement);
};

internal sealed class IndefiniteRetryPolicy : IRetryPolicy
{
    private readonly TimeSpan _delay;
    public IndefiniteRetryPolicy(TimeSpan delay) => _delay = delay;
    public TimeSpan? NextRetryDelay(RetryContext retryContext) => _delay;
}

Reference implementation: (handler registration), (handshake plus session comparison), (BuildConnection), (IndefiniteRetryPolicy).

Method names#

ConnectorMethods declares exactly seven constants. All seven are nameof(...), so the wire string always equals the constant name.

Evidence for the invoke shapes: (GetRange), (Subscribe) and (Unsubscribe), (the four hub methods), (the three pushes).

GetRange, Subscribe and Unsubscribe are not declared on ConnectorHub. They are client-side handlers you register; Atlas reaches them through IHubContext.Clients.Client(connectionId).

Handshake#

Invoking it#

var response = await connection.InvokeAsync<HandshakeResponse>(
    ConnectorMethods.Handshake, advertisement);

The handshake returns a value. Treating it as fire-and-forget compiles and works, and costs you Atlas-restart detection (below).

Call it after StartAsync, again on every Reconnected, and again any time your advertised path set changes. A re-handshake on the same connection is a full replacement: the registry sweeps the connection's prior binding paths and template matchers before applying the new advertisement.

ConnectorAdvertisement#

a session id different from the one it saw on a prior handshake means every subscription id minted before is now unreachable, and the connector tears its local subscription state down.

Without it, after Atlas restarts your connector keeps pushing rows tagged with SubscriptionIds the new process has never minted. Atlas logs "Received rows for unknown subscription {SubscriptionId}; ignoring" at Debug and drops them. That is data loss until someone restarts the connector by hand.

The required behaviour, per the reference implementation:

1. Keep the last-seen AtlasSessionId.
2. On each handshake, compare. If it changed, drop every locally tracked subscription id and run your own Unsubscribe(staleIds) so upstream watches are released.
3. Do not try to re-use the old ids. Atlas reconciles a fresh set on the ConnectorAdvertised event that your handshake just fired.

Serialize concurrent handshakes behind a lock. Startup, Reconnected and a routes-changed re-advertise can all race, and the session comparison plus reset must run as one atomic step.

DatasourceDescriptor#

Note the nullability split exactly. Keys 0-3 are required positional arguments. Keys 4 and 5 are TimeSpan? with = null defaults, which makes them MessagePack-additive: an older deserializer that does not know about them ignores the trailing entries.

Keys 4 and 5 are reserved on this record. They round-trip on the wire and are not read, and none of the shipped datasource descriptors populates them.

Declare latency on SubscriptionAck.AllowedLatency (Key 4) instead. That is a per-binding-path value, recorded at subscribe time and applied by the live-projection gate.

ParameterDescriptor#

`Ingestion/Handshake. ---

Binding paths and templates#

BindingPaths are exact strings, resolved by dictionary lookup. Templates are consulted only when the concrete map misses, by a linear scan of every compiled template in the fleet (the comment there sizes it for "<<100 across the fleet"). Prefer exact paths where you can enumerate them.

Because template resolution is fleet-wide, a GetRange or Subscribe can arrive for a path this datasource never explicitly advertised - only matched by shape. Re-match locally to route it internally and to extract your parameters.

Placeholder grammar#

:

\[(?<name>[A-Za-z_][A-Za-z0-9_]*)(?::(?<type>[A-Za-z_][A-Za-z0-9_]*))?\]

• Two forms: [Name] and [Name:type].
• Square brackets, not curly braces. A template written with braces declares no placeholder at all: it compiles to a literal path that matches only itself.
• A literal that merely looks bracketed is left alone, because a placeholder name must start with a letter or underscore. An IPv6 host [::1] and a JSONPath [0] or [*] pass through untouched.
• The name must start with a letter or underscore.
• A placeholder never spans a /. Untyped placeholders match [^/]+, a single path segment. Multi-segment values are not supported in v1.

Type vocabulary#

and TemplateParameterTypes.MatchPattern:

Tag parsing is case-insensitive (TemplateParameterTypes.TryParse,).

datetime is not available: ISO 8601 contains :, which collides with the type-tag separator. Use a long placeholder carrying ticks or epoch.

Three hard errors#

Each of these throws, and the first two throw inside TemplateCompiler.Compile which runs inside ConnectorHub.Handshake, so the whole handshake fails and your InvokeAsync sees the exception.

Overlap ignores type tags entirely: Overlaps compares the templates with every placeholder replaced by [].

Duplicate paths across connections#

Not an error. The same concrete path advertised by a second connection is last-writer-wins, with a warning: "Binding path '{BindingPath}' was advertised by connector connection '{ExistingConnectionId}' ... and is being overwritten by connection '{NewConnectionId}' ... Last-writer-wins.". There is no fan-out or round-robin anywhere - TryResolve returns exactly one route. Running two instances of the same connector gives you replacement or failover, never load sharing.

GetRange - the pull path#

connection.On<ConnectorRangeRequest, ConnectorRangeResponse>(
    ConnectorMethods.GetRange,
    request => ResolveRange(request));

Atlas invokes it as InvokeAsync<ConnectorRangeResponse>.

IDatasource.GetRange is synchronous and must be thread-safe - the host's SignalR handler runs on the connection's message-loop thread and may be invoked concurrently with other plugin calls.

ConnectorRangeRequest#

Ingestion/ConnectorRequests. ### ConnectorRangeResponse`

PushdownRequest#

Also on the type, and useful: PushdownRequest.Empty and IsEmpty (marked [IgnoreMember] so it does not serialize). The record hand-rolls Equals / GetHashCode over the whole predicate tree so Atlas's per-entity range cache collapses identical requests.

There is no TimeFrom / TimeTo / OrderByTime / Predicates. The time window lives inside the predicate tree as ComparePredicates on a FieldRefExpr against a DateTimeLiteralExpr. See Connector Range Queries for extracting it (TimestampBounds.Extract).

PushdownSortKey:

SortDirection is { Asc, Desc } - not Ascending / Descending. The ordinals are implicit (Asc = 0, Desc = 1) and are still wire-visible, so the declaration order is as load-bearing as any [Key(n)].

PushdownResidual#

. Identical shape to PushdownRequest.

It is a record, not an enum. The two factory members are:

The rule in one line: return the sub-request you did NOT satisfy. Partial answers are the normal case - a historian that applied the time comparisons but nothing else returns a residual whose predicate is the original tree minus those comparisons, with SortKeys / Limit / Skip preserved.

residual: null is illegal. It compiles, it serializes, and it throws a NullReferenceException inside Atlas, where the connector author cannot see it. and both dereference residual.IsEmpty unguarded. Use PushdownResidual.Empty.

PushdownPredicate union#

. The union tags are as wire-breaking as the key numbers.

PushdownCompareOp is { Eq, Neq, Lt, Lte, Gt, Gte }, implicit ordinals 0-5. ### PushdownExpr union

Rowsets and columns#

Both directions carry values the same way, so a plugin writes one rowset builder and reuses it for pull and push.

CompoundRowSet#

Note both details: RowCount is Key 0, and Columns is a Column[] array, not IReadOnlyList<Column>.

RowCount is authoritative, not derived. Atlas reads exactly RowCount cells from every column and never inspects array Length. CompoundRowSetAssembler.Assemble early-returns on rows.RowCount == 0 and iterates for (int r = 0; r < rows.RowCount; r++). A RowCount smaller than your arrays truncates; larger throws IndexOutOfRangeException inside Atlas's ReadCell.

The empty rowset is new CompoundRowSet(0, Array.Empty<Column>()) and.

Because Columns is typed as the abstract base, MessagePack writes the union discriminator per element. No envelope or wrapper type is involved - ConnectorValueEnvelope is gone and was never the mechanism it claimed to be.

The Column union#

. Seven members. Every one is (Key 0 string Name, Key 1 bool[]? Nulls, Key 2 T[] <payload>).

DateTimeColumn's payload is named Ticks, not Values. Convert with new DateTime(ticks, DateTimeKind.Utc) on the way in and value.ToUniversalTime().Ticks on the way out. Always UTC.

Nulls is sparse by convention. Allocate it only when at least one cell is null; pass null otherwise, and never allocate an all-false array. Atlas's IsNullAt treats a null or short Nulls array as "not null". There is no whole-response "null" type. Nullness is per cell, per column. ### Column names are the binding contract

Get the column names right before anything else. They are what binds a returned column to a property field.

Column Name must equal the dotted leaf path of the compound type descriptor the binding is declared against. For the shipped TVQ compound that is exactly:

Timestamp    (DateTimeColumn)
Value        (a column matching the property's declared xsd range)
Quality      (IntColumn)

Not t / v / q. Nested compounds use parent.child.leaf, built by joining the descriptor's leaf path with ..

What happens when a name does not match:

Case matters: "quality" is not "Quality".

Column order is irrelevant on the compound path. Columns are indexed into a name-keyed dictionary before any row is read. The two invariants that do matter are name correctness and arrays at least RowCount long.

One exception where order IS load-bearing. If the bound property's range is a scalar xsd type rather than a compound, BoundColumnWriter.WriteScalar reads rows.Columns[0] positionally and ignores names entirely. A connector that emits the conventional [Timestamp, Value, Quality] rowset against a scalar-ranged property therefore binds the Timestamp. Bind time-series properties to a compound (atlas-common:TVQ), or emit the value column first.

Last row only. On the push path, both WriteScalar and WriteCompound read lastIdx = rows.RowCount - 1 and write just that row into the SoA cell - the v1 write contract. Pushing a 500-row batch updates the cell once, from row 499.

The push-subscription protocol#

This is the only path by which live data reaches Atlas. If you implement nothing in this section, your connector can serve backfill and will never show a current value.

Shape of the exchange#

Atlas → connector   Subscribe(SubscriptionRequest[])  ──InvokeAsync──▶
Atlas ◀── connector   SubscriptionAck[]                (one per request, same order)

Atlas ◀── connector   PushSubscriptionRows(id, rows, Snapshot, asOfT)      x N
Atlas ◀── connector   PushSubscriptionStatus(id, SnapshotComplete, null)
Atlas ◀── connector   PushSubscriptionRows(id, rows, Update, asOfT)        x N
Atlas ◀── connector   PushSubscriptionWatermark(id, watermarkT)            interleaved
        ...
Atlas ◀── connector   PushSubscriptionStatus(id, Faulted|Removed, reason)
   or
Atlas → connector   Unsubscribe(Guid[])               ──SendAsync──▶  (no ack)

Subscribe#

connection.On<IReadOnlyList<SubscriptionRequest>, IReadOnlyList<SubscriptionAck>>(
    ConnectorMethods.Subscribe,
    requests => HandleSubscribe(requests));

The ack list must mirror the request list one-to-one, in order. Atlas pairs them by index, not by id: does var ack = acks[i]; var pending = pendings[i];. A short or reordered list mis-assigns every subscription after the first divergence.

Partial failure in a large batch is normal and must not throw. Surface unsupported paths and invalid pushdowns as SubscriptionStatus.NotSupported in the per-item ack. If the whole invocation throws, Atlas logs "SubscribeAsync threw for route {ConnectionId}/{DatasourceId}; skipping {Count} pending subscription(s)" and abandons the entire batch.

SubscriptionRequest#

What Atlas actually sends today. constructs every SubscriptionRequest with PushdownRequest.Empty. So on a subscription pushdown is always empty in practice, even though the field is fully wired. Handle a non-empty pushdown correctly anyway - the shape is part of the contract and the residual you ack is applied over every subsequent emit. In v1, pushdowns are always PushdownRequest.Empty, and DiagnosticLabel is the bound descriptor's own label when there is a single target, and otherwise "{bindingPath} (x{count})".

SubscriptionAck#

AllowedLatency semantics: your per-binding-path late-arrival tolerance - how far behind wall clock a value for this path may attest and still be waited for. Atlas's full-row live-projection gate reads it per gating series and takes the max across an instance, so one slow path bounds how long an aligned row waits. null means "I declare nothing", and Atlas falls back to the global default LiveProjection:AllowedLatencyMs = 2000 ms (range 0 to 86 400 000, validated at startup -). A null on a later ack also clears a prior declaration for that path.

It is declared per path rather than per datasource precisely because fast and slow paths share one datasource.

SubscriptionStatus#

. Explicit numeric values - they are the wire encoding.

On a terminal status Atlas logs Subscription {SubscriptionId} on {BindingPath} terminated ({Status}) ... cells retain last value. Reason: {Reason}.

The watermark reset on a terminal status is what lets the next subscription's snapshot rows through; without it they would be treated as late.

EmitKind#

. Explicit values.

For a pure tick stream the distinction is cosmetic. For rowset-shaped subscriptions (a SQL query, say) it is load-bearing for correct state reconstruction.

Refresh rule: a Snapshot emit arriving after a SnapshotComplete is a refresh signal. The consumer drops accumulated state and treats the new Snapshot rows as authoritative until the next SnapshotComplete. Atlas's v1 consumer handles this transparently because last-row-only SoA writes overwrite anyway.

The three push methods#

All three are positional hub-method parameters, not DTOs. There is no wrapper record. Order and type must match exactly:

PushSubscriptionRows(Guid subscriptionId, CompoundRowSet rows, EmitKind kind, long asOfT)
PushSubscriptionStatus(Guid subscriptionId, SubscriptionStatus status, string? reason)
PushSubscriptionWatermark(Guid subscriptionId, long watermarkT)

Send them with SendAsync (fire and forget), the way the shipped sink does:

connection.SendAsync(ConnectorMethods.PushSubscriptionRows,
                     subscriptionId, rows, kind, asOfT);

InvokeAsync would work on the wire but blocks a push loop for no benefit - Atlas returns Task.CompletedTask from all three.

Emits sent while the connection is not Connected are dropped, not queued. The shipped sink checks connection.State != HubConnectionState.Connected and logs at Debug. Treat delivery as best-effort and rely on the Faulted / re-subscribe lifecycle for recovery, never on guaranteed delivery of every tick.

Unknown subscription ids are ignored at Debug on all three paths. This is why AtlasSessionId matters.

asOfT and watermarkT#

Both are UTC ticks on the plugin's own clock. Both are long. They mean different things.

Choosing asOfT:

• Polled plugins: poll_as_of − allowedLatency.
• Realtime-push plugins: the source event timestamp.

Emitting watermarks:

• Polled plugins: emit after every poll, including polls that produced no rows. Downstream calculated properties are watermark-gated and will freeze on a quiet tag otherwise.
• Realtime-push plugins: run a heartbeat timer and emit watermarkT = wall_clock_now − allowedLatency when no value event has fired since the last heartbeat. The shipped helper is ConnectorHeartbeatScheduler (Timebase.Connect.Abstractions/), whose tick granularity is 100 ms by default and which computes exactly nowTicks - entry.AllowedLatency.Ticks.

Watermarks must never regress for a given SubscriptionId. That monotonicity is the plugin's responsibility; nothing on the wire or in Atlas enforces it.

Unsubscribe#

connection.On<IReadOnlyList<Guid>>(
    ConnectorMethods.Unsubscribe,
    ids => HandleUnsubscribe(ids));

Atlas sends this with SendAsync - fire and forget, no ack, no return value. The contract is "best-effort no-op on unknown ids": an id may already have faulted, been removed, or never existed. After the call returns you must not emit further rows or status for those ids.

Full C# model definitions#

Verbatim from the shared DTO assembly. Namespaces: Timebase.Atlas.Shared.Ingestion, Timebase.Atlas.Shared.Pushdown, Timebase.Atlas.Shared.Templates.

// Ingestion: ConnectorMethods
public static class ConnectorMethods
{
    public const string Handshake                 = nameof(Handshake);
    public const string GetRange                  = nameof(GetRange);
    public const string Subscribe                 = nameof(Subscribe);
    public const string Unsubscribe               = nameof(Unsubscribe);
    public const string PushSubscriptionRows      = nameof(PushSubscriptionRows);
    public const string PushSubscriptionStatus    = nameof(PushSubscriptionStatus);
    public const string PushSubscriptionWatermark = nameof(PushSubscriptionWatermark);
}

// Ingestion: Handshake
[MessagePackObject]
public sealed record ConnectorAdvertisement(
    [property: Key(0)] string ConnectorId,
    [property: Key(1)] string DisplayName,
    [property: Key(2)] IReadOnlyList<DatasourceDescriptor> Datasources);

[MessagePackObject]
public sealed record HandshakeResponse(
    [property: Key(0)] Guid AtlasSessionId);

[MessagePackObject]
public sealed record DatasourceDescriptor(
    [property: Key(0)] string DatasourceId,
    [property: Key(1)] IReadOnlyList<ParameterDescriptor> Parameters,
    [property: Key(2)] IReadOnlyList<string> BindingPaths,
    [property: Key(3)] IReadOnlyList<string> Templates,
    [property: Key(4)] TimeSpan? AllowedLatency = null,      // reserved
    [property: Key(5)] TimeSpan? HeartbeatInterval = null);  // reserved

[MessagePackObject]
public sealed record ParameterDescriptor(
    [property: Key(0)] string Name,
    [property: Key(1)] string Type,
    [property: Key(2)] bool Required);

// Ingestion: ConnectorRequests
[MessagePackObject]
public sealed record ConnectorRangeRequest(
    [property: Key(0)] string BindingPath,
    [property: Key(1)] PushdownRequest Pushdown);

// Ingestion: ConnectorResponses
[MessagePackObject]
public sealed record CompoundRowSet(
    [property: Key(0)] int RowCount,
    [property: Key(1)] Column[] Columns);

[MessagePackObject]
public sealed record ConnectorRangeResponse(
    [property: Key(0)] CompoundRowSet Rows,
    [property: Key(1)] PushdownResidual Residual);

// Ingestion: Columns
[Union(0, typeof(DoubleColumn))]
[Union(1, typeof(LongColumn))]
[Union(2, typeof(StringColumn))]
[Union(3, typeof(BoolColumn))]
[Union(4, typeof(DateTimeColumn))]
[Union(5, typeof(FloatColumn))]
[Union(6, typeof(IntColumn))]
public abstract record Column;

[MessagePackObject]
public sealed record DoubleColumn(
    [property: Key(0)] string Name,
    [property: Key(1)] bool[]? Nulls,
    [property: Key(2)] double[] Values) : Column;
// LongColumn(long[] Values), StringColumn(string[] Values),
// BoolColumn(bool[] Values), FloatColumn(float[] Values),
// IntColumn(int[] Values) follow the identical (Name, Nulls, Values) layout.

/// <summary>UTC ticks (matches System.DateTime.Ticks).</summary>
[MessagePackObject]
public sealed record DateTimeColumn(
    [property: Key(0)] string Name,
    [property: Key(1)] bool[]? Nulls,
    [property: Key(2)] long[] Ticks) : Column;   // note: Ticks, not Values

// Ingestion: Subscriptions
[MessagePackObject]
public sealed record SubscriptionRequest(
    [property: Key(0)] Guid SubscriptionId,
    [property: Key(1)] string BindingPath,
    [property: Key(2)] PushdownRequest Pushdown,
    [property: Key(3)] string? DiagnosticLabel);

[MessagePackObject]
public sealed record SubscriptionAck(
    [property: Key(0)] Guid SubscriptionId,
    [property: Key(1)] SubscriptionStatus Status,
    [property: Key(2)] PushdownResidual Residual,
    [property: Key(3)] string? Error,
    [property: Key(4)] TimeSpan? AllowedLatency = null);

public enum SubscriptionStatus
{
    Active = 0, Pending = 1, NotSupported = 2,
    SnapshotComplete = 3, Faulted = 4, Removed = 5,
}

public enum EmitKind { Snapshot = 0, Update = 1 }

// Pushdown: PushdownRequest
public enum SortDirection { Asc, Desc }

[MessagePackObject]
public sealed record PushdownSortKey(
    [property: Key(0)] FieldRefExpr Field,
    [property: Key(1)] SortDirection Direction);

[MessagePackObject]
public sealed record PushdownRequest(
    [property: Key(0)] PushdownPredicate? Predicate,
    [property: Key(1)] IReadOnlyList<PushdownSortKey> SortKeys,
    [property: Key(2)] int? Limit,
    [property: Key(3)] int? Skip);   // + static Empty, + IsEmpty [IgnoreMember]

// Pushdown: PushdownResidual
[MessagePackObject]
public sealed record PushdownResidual(
    [property: Key(0)] PushdownPredicate? Predicate,
    [property: Key(1)] IReadOnlyList<PushdownSortKey> SortKeys,
    [property: Key(2)] int? Limit,
    [property: Key(3)] int? Skip)
{
    public static PushdownResidual From(PushdownRequest request);  // satisfied nothing
    public static PushdownResidual Empty { get; }                  // satisfied everything
    [IgnoreMember] public bool IsEmpty { get; }
}

// Pushdown: PushdownPredicate
[Union(0, typeof(AndPredicate))]
[Union(1, typeof(OrPredicate))]
[Union(2, typeof(NotPredicate))]
[Union(3, typeof(ComparePredicate))]
public abstract record PushdownPredicate;

public enum PushdownCompareOp { Eq, Neq, Lt, Lte, Gt, Gte }

[MessagePackObject]
public sealed record ComparePredicate(
    [property: Key(0)] PushdownCompareOp Op,
    [property: Key(1)] PushdownExpr Left,
    [property: Key(2)] PushdownExpr Right) : PushdownPredicate;

[Union(0, typeof(FieldRefExpr))]
[Union(1, typeof(DoubleLiteralExpr))]
[Union(2, typeof(LongLiteralExpr))]
[Union(3, typeof(StringLiteralExpr))]
[Union(4, typeof(BoolLiteralExpr))]
[Union(5, typeof(DateTimeLiteralExpr))]
[Union(6, typeof(NullLiteralExpr))]
public abstract record PushdownExpr;

[MessagePackObject]
public sealed record FieldRefExpr(
    [property: Key(0)] IReadOnlyList<string> Path) : PushdownExpr;

/// <summary>UTC ticks (matches System.DateTime.Ticks).</summary>
[MessagePackObject]
public sealed record DateTimeLiteralExpr([property: Key(0)] long Ticks) : PushdownExpr;

Getting the types into your project#

There is no published NuGet package. is eleven lines and declares only TargetFramework (net10.0), Nullable, ImplicitUsings and one PackageReference to MessagePack 3.1.4. There is no PackageId, no Version, no IsPackable, no GeneratePackageOnBuild and no .nuspec. All twelve in-repo consumers use ProjectReference. A <PackageReference Include="Timebase.Atlas.Shared.DTOs" /> fails restore with NU1101 on the very first step.

In-repo connector (the shipped pattern):

<ProjectReference Include="..\..\..\Timebase.Atlas\src\shared\Timebase.Atlas.Shared.DTOs\Timebase.Atlas.Shared.DTOs.csproj" />

Out-of-tree connector: redeclare the records in your own assembly with the exact [Key(n)] numbers, [Union(n)] tags and enum ordinals from this article, and add <PackageReference Include="MessagePack" Version="3.1.4" />. You must also reproduce ConnectorMessagePackOptions.Options yourself:

public static readonly IFormatterResolver Resolver =
    CompositeResolver.Create(NativeGuidResolver.Instance, StandardResolver.Instance);

public static readonly MessagePackSerializerOptions Options =
    MessagePackSerializerOptions.Standard
        .WithResolver(Resolver)
        .WithCompression(MessagePackCompression.Lz4BlockArray);

This is why every key number, union tag and enum ordinal in this article is load-bearing. A redeclaration that drifts from Atlas by one index does not fail to compile.

Sixteen things to get right#

Transport notes#

• No receive size cap on Atlas's side. The default 32 KB limit is removed, because a connector returning a wide CompoundRowSet exceeds it immediately. The request body limit is unlimited for the same reason, and is overridable through Kestrel:Limits:MaxRequestBodySize.
• Raise HubConnection.ServerTimeout. The shipped client uses 5 minutes so a long-polling transport carrying a wide ConnectorRangeResponse does not trip the default 30-second idle timeout mid-transfer.
• MessagePack only. No JSON protocol is registered on this hub. transport, so WebSockets and long polling are both reachable, but no code pins a preferred one.