mirror of
https://github.com/HeyPuter/puter.git
synced 2026-09-20 20:26:21 +00:00
feat: pending event deliveries and delivery-class invariants (PUT-1676) (#3680)
* feat: pending event deliveries and delivery-class invariants (PUT-1676)
* fix: make pending delivery claims and drains atomic, keep the region under its ceiling (PUT-1676)
- claim() and the drain-time reindex run as Lua over the subscription's own
{subId}-tagged keys. Two claimers can no longer both lease the head, and a
drain that finds the queue empty deletes it in the same step it checks, so a
concurrent enqueue is never wiped between the two.
- An append writes the entry and its queue position in one MULTI (same slot),
with the index seeded before it and corrected after, so an entry is never
visible without its position and never left out of the sweeper's index.
- Pipelines no longer mix slots (index/counter vs. per-subscription keys), so
the store works on a multi-shard cluster, not only a single-shard one.
- Region shedding counts the marker it leaves behind; it used to stop one over
the ceiling and convert a real event into a marker on every enqueue after.
- A claimed or suspended subscription moves to the back of the sweeper's index,
so a delivery nobody settles cannot hold the head against every other backlog.
- `single` rows must carry a `worker` target: with sockets exhausted and no
handler, an unacknowledged delivery would sit at the head forever.
- A gap marker for a row with no socket target is dropped rather than counted
as a delivery of nothing.
- Backlog keys carry a 7-day TTL, refreshed by every claim, as a backstop for
keys a purge/enqueue race left unindexed.
This commit is contained in:
Daniel Salazar
@@ -75,6 +75,15 @@ export const EVENTS_SUBSCRIBE_LIMIT = userWindow('events:subscribe', 60);
|
||||
*/
|
||||
export const EVENTS_LIST_LIMIT = userWindow('events:list', 120);
|
||||
|
||||
/**
|
||||
* Delivery acknowledgements per minute, per user.
|
||||
*
|
||||
* One ack per `single` delivery, so this sits level with what one subscription
|
||||
* may be delivered — a client acking faster than that is acking things it was
|
||||
* never sent.
|
||||
*/
|
||||
export const EVENTS_ACK_LIMIT = userWindow('events:ack', 600);
|
||||
|
||||
// -- Dispatch fan-out ------------------------------------------------
|
||||
|
||||
/**
|
||||
@@ -103,6 +112,29 @@ export const EVENTS_BROADCAST_DELIVERY_LIMIT = userWindow(
|
||||
*/
|
||||
export { FILTER_EVALUATIONS_PER_EVENT } from '../../services/events/matcher.js';
|
||||
|
||||
// -- Undelivered backlog ---------------------------------------------
|
||||
|
||||
/**
|
||||
* Deliveries one subscription may hold undelivered.
|
||||
*
|
||||
* A `single` delivery waits until something takes it, so a subscription whose
|
||||
* consumer is gone accumulates. Over the cap the oldest go and one gap marker
|
||||
* takes their place, which is what keeps "at-least-once" honest: what was lost
|
||||
* is visible rather than silent.
|
||||
*/
|
||||
export const EVENTS_PENDING_DELIVERIES_PER_SUBSCRIPTION = 10_000;
|
||||
|
||||
/**
|
||||
* Undelivered deliveries one region may hold across every subscription.
|
||||
*
|
||||
* The per-subscription cap bounds one backlog and nothing in aggregate —
|
||||
* multiply it by the subscriptions that can exist and the region's memory is
|
||||
* the only remaining limit. Over this, the oldest deliveries in the region are
|
||||
* shed first, each shedding subscription gets a gap marker, and an alarm says
|
||||
* it happened.
|
||||
*/
|
||||
export const EVENTS_REGION_PENDING_CEILING = 1_000_000;
|
||||
|
||||
// -- Coalescing ------------------------------------------------------
|
||||
|
||||
/**
|
||||
|
||||
@@ -35,6 +35,7 @@ import type { FSEntry } from '../../stores/fs/FSEntry.js';
|
||||
import type { IConfig } from '../../types.js';
|
||||
import {
|
||||
EventsService,
|
||||
EVENTS_ACK_VERB,
|
||||
EVENTS_DELIVERY_CHANNEL,
|
||||
EVENTS_SUBSCRIBE_VERB,
|
||||
EVENTS_UNSUBSCRIBE_VERB,
|
||||
@@ -183,6 +184,7 @@ const appStore = {
|
||||
*/
|
||||
const durableSubscriptionStore = {
|
||||
warmRegion: async () => false,
|
||||
getBySubId: async () => null,
|
||||
};
|
||||
|
||||
/**
|
||||
@@ -383,9 +385,27 @@ describe('subscribing', () => {
|
||||
anchor: { uid: documents.uid, path: documents.path },
|
||||
match: null,
|
||||
op: null,
|
||||
// The connection is the only thing a session row can be delivered
|
||||
// to, so it is the only transport it can ask for.
|
||||
targets: ['socket'],
|
||||
});
|
||||
});
|
||||
|
||||
it('refuses a session subscription a connection could not carry', async () => {
|
||||
seedTree();
|
||||
|
||||
for (const targets of [['worker'], ['socket', 'push']])
|
||||
await expect(
|
||||
service.subscribe(actorFor(), socketId, {
|
||||
subject: `fs:/u${userId}/Documents`,
|
||||
targets,
|
||||
}),
|
||||
).rejects.toSatisfy(
|
||||
(err: unknown) =>
|
||||
isHttpError(err) && err.legacyCode === 'invalid_targets',
|
||||
);
|
||||
});
|
||||
|
||||
it('rejects a subject that resolves to nothing', async () => {
|
||||
await expect(subscribe('fs:/nowhere/at/all')).rejects.toSatisfy(
|
||||
(err: unknown) =>
|
||||
@@ -1154,6 +1174,26 @@ describe('the socket surface', () => {
|
||||
expect(ack.mock.calls[0][0]).toEqual({ ok: true });
|
||||
});
|
||||
|
||||
it('answers an ack for a subscription the caller does not hold', async () => {
|
||||
const socket = fakeSocket();
|
||||
service.attachSocket(socket, actorFor());
|
||||
|
||||
const ack = vi.fn();
|
||||
socket.fire(
|
||||
EVENTS_ACK_VERB,
|
||||
{ subId: 'app#someone-elses', id: '1-0' },
|
||||
ack,
|
||||
);
|
||||
await vi.waitFor(() => expect(ack).toHaveBeenCalled());
|
||||
|
||||
expect(ack.mock.calls[0][0]).toEqual({
|
||||
ok: false,
|
||||
error: expect.objectContaining({
|
||||
code: 'subscription_does_not_exist',
|
||||
}),
|
||||
});
|
||||
});
|
||||
|
||||
it('addresses deliveries at the socket that asked for them', async () => {
|
||||
vi.useFakeTimers();
|
||||
const { documents, file } = seedTree();
|
||||
|
||||
@@ -20,6 +20,7 @@
|
||||
import { randomUUID } from 'node:crypto';
|
||||
import type { EventKey } from '../../clients/event/types.js';
|
||||
import {
|
||||
EVENTS_ACK_LIMIT,
|
||||
EVENTS_BROADCAST_DELIVERY_LIMIT,
|
||||
EVENTS_COALESCE_WINDOW_MS,
|
||||
EVENTS_MATCHED_SUBSCRIPTIONS_PER_EVENT,
|
||||
@@ -35,8 +36,15 @@ import {
|
||||
type GenerationBump,
|
||||
type SessionSubscription,
|
||||
} from '../../stores/events/EventSubscriptionStore.js';
|
||||
import type {
|
||||
ClaimedDelivery,
|
||||
PendingShed,
|
||||
} from '../../stores/events/PendingDeliveryStore.js';
|
||||
import {
|
||||
DEFAULT_DURABLE_TARGETS,
|
||||
SESSION_TARGETS,
|
||||
isSubscriptionTarget,
|
||||
targetsAllowedForDelivery,
|
||||
type SubscriptionTarget,
|
||||
} from '../../stores/events/types.js';
|
||||
import type { FSEntry } from '../../stores/fs/FSEntry.js';
|
||||
@@ -66,13 +74,21 @@ import {
|
||||
} from './matcher.js';
|
||||
import {
|
||||
lookupPublicSubject,
|
||||
type DeliverableEvent,
|
||||
type DeliveryClass,
|
||||
type EventContext,
|
||||
type GapMarker,
|
||||
type GapReason,
|
||||
type ProjectedEvent,
|
||||
type PublicSubject,
|
||||
} from './registry.js';
|
||||
import { SubscriptionCache } from './subscriptionCache.js';
|
||||
import { parseSubject, type FsOp } from './subjects.js';
|
||||
import {
|
||||
RecordingWorkerInvoker,
|
||||
type WorkerInvocation,
|
||||
type WorkerInvokerSeam,
|
||||
} from './workerSeam.js';
|
||||
|
||||
/**
|
||||
* Subscribe, unsubscribe, and the dispatch hot path.
|
||||
@@ -100,12 +116,19 @@ import { parseSubject, type FsOp } from './subjects.js';
|
||||
|
||||
export interface SubscribeRequest {
|
||||
subject?: unknown;
|
||||
targets?: unknown;
|
||||
}
|
||||
|
||||
export interface UnsubscribeRequest {
|
||||
subId?: unknown;
|
||||
}
|
||||
|
||||
/** Body of the `events.ack` verb: which subscription, and which delivery. */
|
||||
export interface AckRequest {
|
||||
subId?: unknown;
|
||||
id?: unknown;
|
||||
}
|
||||
|
||||
/** Body of `POST /events/subscribe`. */
|
||||
export interface DurableSubscribeRequest extends SubscribeRequest {
|
||||
delivery?: unknown;
|
||||
@@ -128,6 +151,7 @@ export interface SubscriptionView {
|
||||
anchor: { uid: string; path: string };
|
||||
match: string | null;
|
||||
op: FsOp | null;
|
||||
targets: SubscriptionTarget[];
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -137,7 +161,6 @@ export interface SubscriptionView {
|
||||
*/
|
||||
export interface DurableSubscriptionView extends SubscriptionView {
|
||||
delivery: DeliveryClass;
|
||||
targets: SubscriptionTarget[];
|
||||
handlerName: string | null;
|
||||
appUid: string | null;
|
||||
createdAt: number;
|
||||
@@ -150,35 +173,28 @@ export type VerbAck<T extends object> =
|
||||
| ({ ok: true } & T)
|
||||
| { ok: false; error: { code: string; message: string } };
|
||||
|
||||
export type GapReason =
|
||||
| 'matched_subscription_limit'
|
||||
| 'filter_evaluation_limit'
|
||||
| 'delivery_rate_limit';
|
||||
|
||||
/**
|
||||
* `gap` says an event existed and was not delivered. It rides the delivery
|
||||
* channel because a subscriber that never saw one would read the silence as
|
||||
* "nothing happened", and carries no `uid`/`path` — what was dropped is exactly
|
||||
* what it cannot name.
|
||||
*/
|
||||
export interface GapMarker {
|
||||
id: string;
|
||||
subject: string;
|
||||
op: 'gap';
|
||||
reason: GapReason;
|
||||
ts: number;
|
||||
}
|
||||
export type { GapMarker, GapReason } from './registry.js';
|
||||
|
||||
/** One delivery, as the client receives it. */
|
||||
export interface DeliveryEnvelope {
|
||||
subId: string;
|
||||
event: ProjectedEvent | GapMarker;
|
||||
event: DeliverableEvent;
|
||||
/** Set on a `single`: nothing else takes this until `events.ack` settles it. */
|
||||
ackRequired?: true;
|
||||
/** The handle `events.ack` names — the delivery, not the event. */
|
||||
ackId?: string;
|
||||
}
|
||||
|
||||
/** The envelope plus where it goes. The address is not part of the wire. */
|
||||
/**
|
||||
* The envelope plus where it goes. The address is not part of the wire, and
|
||||
* neither is the handler a durable row may want run alongside the socket copy.
|
||||
*/
|
||||
interface AddressedDelivery {
|
||||
target: SocketSpecifier;
|
||||
envelope: DeliveryEnvelope;
|
||||
/** False for a row that asked for its handler and no socket copy. */
|
||||
socket: boolean;
|
||||
worker?: WorkerInvocation;
|
||||
}
|
||||
|
||||
/** What a dispatch call site can supply that the event itself does not carry. */
|
||||
@@ -198,6 +214,7 @@ export interface FsDispatchOptions {
|
||||
|
||||
export const EVENTS_SUBSCRIBE_VERB = 'events.subscribe';
|
||||
export const EVENTS_UNSUBSCRIBE_VERB = 'events.unsubscribe';
|
||||
export const EVENTS_ACK_VERB = 'events.ack';
|
||||
export const EVENTS_DELIVERY_CHANNEL = 'events.delivery';
|
||||
|
||||
// -- Expiry sweep -----------------------------------------------------
|
||||
@@ -212,6 +229,27 @@ const EXPIRY_BATCH_SIZE = 500;
|
||||
/** Batches one sweep takes, so a large backlog drains over several passes. */
|
||||
const EXPIRY_MAX_BATCHES = 50;
|
||||
|
||||
// -- Owed deliveries --------------------------------------------------
|
||||
|
||||
/**
|
||||
* Socket attempts one `single` delivery gets before its handler takes it. Two,
|
||||
* because the second is what a client that reconnected — or a second connection
|
||||
* of the same account — is worth trying; a third is just a slower hand-off.
|
||||
*/
|
||||
const SINGLE_SOCKET_ATTEMPTS = 2;
|
||||
|
||||
/** How often expired leases are reclaimed and owed deliveries retried. */
|
||||
const PENDING_SWEEP_INTERVAL_MS = 10_000;
|
||||
|
||||
/** Subscriptions one sweep pass looks at, taken from the oldest end. */
|
||||
const PENDING_SWEEP_SUBSCRIPTIONS = 100;
|
||||
|
||||
/**
|
||||
* Deliveries one subscription may be handed in a row. A consumer that settles
|
||||
* inline would otherwise drain a whole backlog inside one ack.
|
||||
*/
|
||||
const PENDING_DRAIN_BATCH = 25;
|
||||
|
||||
/** The part of a socket this service uses, so tests need not build one. */
|
||||
export interface EventSocket {
|
||||
id: string;
|
||||
@@ -236,10 +274,9 @@ const tooManyCalls = (): HttpError =>
|
||||
legacyCode: 'too_many_requests',
|
||||
});
|
||||
|
||||
/** Stands until there is a pending-delivery store to take a `single` lease. */
|
||||
const deliveryClassUnavailable = (): HttpError =>
|
||||
new HttpError(501, 'Delivery class `single` is not available yet', {
|
||||
legacyCode: 'delivery_class_unavailable',
|
||||
const handlerRequired = (): HttpError =>
|
||||
new HttpError(400, 'A `single` subscription needs a handlerName', {
|
||||
legacyCode: 'events_handler_required',
|
||||
});
|
||||
|
||||
const badRequest = (message: string, code: string): HttpError =>
|
||||
@@ -266,12 +303,12 @@ const toView = (sub: DispatchSubscription): SubscriptionView => ({
|
||||
anchor: { uid: sub.anchorUid, path: sub.anchorPath },
|
||||
match: sub.match,
|
||||
op: sub.op,
|
||||
targets: sub.targets ?? SESSION_TARGETS,
|
||||
});
|
||||
|
||||
const toDurableView = (sub: DurableSubscription): DurableSubscriptionView => ({
|
||||
...toView(sub),
|
||||
delivery: sub.delivery,
|
||||
targets: sub.targets,
|
||||
handlerName: sub.handlerName,
|
||||
appUid: sub.appUid,
|
||||
createdAt: sub.createdAt,
|
||||
@@ -300,10 +337,17 @@ const deliveryTarget = (row: DispatchSubscription): SocketSpecifier => {
|
||||
};
|
||||
};
|
||||
|
||||
/** Transports a row is asking for, whichever store it came from. */
|
||||
const targetsOf = (row: DispatchSubscription): SubscriptionTarget[] =>
|
||||
row.durable === true
|
||||
? (row.targets ?? DEFAULT_DURABLE_TARGETS)
|
||||
: SESSION_TARGETS;
|
||||
|
||||
/**
|
||||
* Rows this pass can actually deliver. Durable `single` rows need the pending
|
||||
* store to take a lease, and a row with no socket target has asked not to be
|
||||
* delivered over one.
|
||||
* Whether this pass has anywhere to put the row. A `single` is queued whether
|
||||
* or not anything is listening right now — that is what it is for — while a
|
||||
* `broadcast` whose only transport is one this build cannot carry has no
|
||||
* target, and an event with no target is dropped rather than held.
|
||||
*/
|
||||
const nowSeconds = (): number => Math.floor(Date.now() / 1000);
|
||||
|
||||
@@ -314,30 +358,33 @@ const unexpired = (row: DispatchSubscription): boolean => {
|
||||
return expiresAt === null || expiresAt > nowSeconds();
|
||||
};
|
||||
|
||||
const deliverableOverSockets = (row: DispatchSubscription): boolean =>
|
||||
unexpired(row) &&
|
||||
(row.durable !== true ||
|
||||
(row.delivery === 'broadcast' &&
|
||||
(row.targets ?? []).includes('socket')));
|
||||
const deliverable = (row: DispatchSubscription): boolean => {
|
||||
if (row.durable !== true) return true;
|
||||
if (!unexpired(row)) return false;
|
||||
if (row.delivery === 'single') return true;
|
||||
const targets = targetsOf(row);
|
||||
return targets.includes('socket') || targets.includes('worker');
|
||||
};
|
||||
|
||||
const isSingle = (row: DispatchSubscription): boolean =>
|
||||
row.durable === true && row.delivery === 'single';
|
||||
|
||||
// -- Durable request parsing ------------------------------------------
|
||||
|
||||
/** Transports a durable row takes unless the caller says otherwise. */
|
||||
const DEFAULT_DURABLE_TARGETS: SubscriptionTarget[] = ['socket', 'worker'];
|
||||
|
||||
/** Longest a `handlerName` may be, matching the column that holds it. */
|
||||
const HANDLER_NAME_MAX_LENGTH = 128;
|
||||
|
||||
const parseDelivery = (value: unknown): DeliveryClass => {
|
||||
if (value === undefined || value === null || value === 'broadcast')
|
||||
return 'broadcast';
|
||||
// Creatable but inert is worse than refused: a `single` row would take a
|
||||
// lease nothing in this build can settle.
|
||||
if (value === 'single') throw deliveryClassUnavailable();
|
||||
if (value === 'single') return 'single';
|
||||
throw badRequest(`Unknown delivery class: ${String(value)}`, 'bad_request');
|
||||
};
|
||||
|
||||
const parseTargets = (value: unknown): SubscriptionTarget[] => {
|
||||
const parseTargets = (
|
||||
value: unknown,
|
||||
delivery: DeliveryClass,
|
||||
): SubscriptionTarget[] => {
|
||||
if (value === undefined || value === null) return DEFAULT_DURABLE_TARGETS;
|
||||
if (!Array.isArray(value) || value.length === 0)
|
||||
throw badRequest(
|
||||
@@ -346,7 +393,34 @@ const parseTargets = (value: unknown): SubscriptionTarget[] => {
|
||||
);
|
||||
if (!value.every(isSubscriptionTarget))
|
||||
throw badRequest('Unknown delivery target', 'invalid_targets');
|
||||
return [...new Set(value)];
|
||||
|
||||
const targets = [...new Set(value)];
|
||||
if (!targetsAllowedForDelivery(delivery, targets))
|
||||
throw badRequest(
|
||||
'A `single` subscription needs a `worker` target and may not target `push`',
|
||||
'invalid_targets',
|
||||
);
|
||||
return targets;
|
||||
};
|
||||
|
||||
/**
|
||||
* A session row is one connection, so the socket is the only transport it can
|
||||
* have: a handler runs long after the connection is gone, and a device
|
||||
* notification is not addressed to a connection at all.
|
||||
*/
|
||||
const parseSessionTargets = (value: unknown): SubscriptionTarget[] => {
|
||||
if (value === undefined || value === null) return SESSION_TARGETS;
|
||||
if (!Array.isArray(value) || value.length === 0)
|
||||
throw badRequest(
|
||||
'targets must be a non-empty array',
|
||||
'invalid_targets',
|
||||
);
|
||||
if (!value.every((target) => target === 'socket'))
|
||||
throw badRequest(
|
||||
'A session subscription may only target `socket`',
|
||||
'invalid_targets',
|
||||
);
|
||||
return SESSION_TARGETS;
|
||||
};
|
||||
|
||||
const parseHandlerName = (value: unknown): string | null => {
|
||||
@@ -394,6 +468,14 @@ export class EventsService extends PuterService {
|
||||
#coalescer: DeliveryCoalescer<AddressedDelivery> | null = null;
|
||||
#expirySweep: ReturnType<typeof setInterval> | null = null;
|
||||
#expiryKick: ReturnType<typeof setTimeout> | null = null;
|
||||
#pendingSweep: ReturnType<typeof setInterval> | null = null;
|
||||
|
||||
/**
|
||||
* What runs an app's handler. The default records the intent and settles
|
||||
* nothing, so a delivery handed to it stays owed until there is a real
|
||||
* invoker to take it.
|
||||
*/
|
||||
worker: WorkerInvokerSeam = new RecordingWorkerInvoker();
|
||||
|
||||
// -- Lifecycle ---------------------------------------------------
|
||||
|
||||
@@ -413,6 +495,7 @@ export class EventsService extends PuterService {
|
||||
},
|
||||
);
|
||||
this.#armExpirySweep();
|
||||
this.#armPendingSweep();
|
||||
}
|
||||
|
||||
override onServerPrepareShutdown(): void {
|
||||
@@ -420,6 +503,8 @@ export class EventsService extends PuterService {
|
||||
this.#expiryKick = null;
|
||||
if (this.#expirySweep) clearInterval(this.#expirySweep);
|
||||
this.#expirySweep = null;
|
||||
if (this.#pendingSweep) clearInterval(this.#pendingSweep);
|
||||
this.#pendingSweep = null;
|
||||
}
|
||||
|
||||
override onServerShutdown(): void {
|
||||
@@ -462,6 +547,13 @@ export class EventsService extends PuterService {
|
||||
});
|
||||
}) as (...args: never[]) => void);
|
||||
|
||||
socket.on(EVENTS_ACK_VERB, ((payload: AckRequest, ack: unknown) => {
|
||||
void this.#answer(ack, async () => {
|
||||
await this.ackDelivery(actor, payload);
|
||||
return {};
|
||||
});
|
||||
}) as (...args: never[]) => void);
|
||||
|
||||
socket.once('disconnect', (() => {
|
||||
void this.reapSocket(userId, socket.id);
|
||||
}) as (...args: never[]) => void);
|
||||
@@ -500,6 +592,7 @@ export class EventsService extends PuterService {
|
||||
|
||||
await this.#spendCallBudget(holderUserId);
|
||||
|
||||
const targets = parseSessionTargets(request?.targets);
|
||||
const rawSubject = String(request?.subject ?? '');
|
||||
const anchor = await this.#resolveSubscribeAnchor(actor, rawSubject);
|
||||
|
||||
@@ -516,6 +609,7 @@ export class EventsService extends PuterService {
|
||||
op: anchor.op,
|
||||
appUid: actor.effectiveApp?.uid ?? null,
|
||||
permission: anchor.permission,
|
||||
targets,
|
||||
};
|
||||
|
||||
const bump = await this.stores.eventSubscription.add(sub);
|
||||
@@ -587,11 +681,16 @@ export class EventsService extends PuterService {
|
||||
await this.#spendCallBudget(holderUserId);
|
||||
|
||||
const delivery = parseDelivery(request?.delivery);
|
||||
const targets = parseTargets(request?.targets);
|
||||
const targets = parseTargets(request?.targets, delivery);
|
||||
const handlerName = parseHandlerName(request?.handlerName);
|
||||
const context = parseContext(request?.context);
|
||||
const expiresAt = parseExpiresAt(request?.expiresAt);
|
||||
|
||||
// A `single` is owed to exactly one consumer, and the handler is the
|
||||
// only one that is always there to take it. Whether the handler exists
|
||||
// is the publish surface's question, not this one's.
|
||||
if (delivery === 'single' && !handlerName) throw handlerRequired();
|
||||
|
||||
const rawSubject = String(request?.subject ?? '');
|
||||
const anchor = await this.#resolveSubscribeAnchor(actor, rawSubject);
|
||||
|
||||
@@ -676,10 +775,47 @@ export class EventsService extends PuterService {
|
||||
throw unknownSubscription();
|
||||
|
||||
const bump = await this.stores.durableSubscription.remove(row);
|
||||
// Whatever it was still owed goes with it: a backlog held for a
|
||||
// subscription nobody can consume is memory, and the paths it names are
|
||||
// ones its holder just gave up asking about.
|
||||
await this.stores.pendingDelivery.purge(subId);
|
||||
this.#forget(subId);
|
||||
this.#publishGeneration(bump, true);
|
||||
}
|
||||
|
||||
/**
|
||||
* Settle a `single` delivery the client took. An id this subscription is
|
||||
* not holding — one already settled, or one reclaimed and handed on — is
|
||||
* not an error: at-least-once makes a duplicate ack routine.
|
||||
*/
|
||||
async ackDelivery(actor: Actor, request: AckRequest): Promise<void> {
|
||||
if (!this.enabled) throw disabled();
|
||||
const holderUserId = actor.user?.id;
|
||||
if (holderUserId === undefined) throw disabled();
|
||||
|
||||
const subId = String(request?.subId ?? '');
|
||||
const entryId = String(request?.id ?? '');
|
||||
if (!subId || !entryId) throw unknownSubscription();
|
||||
|
||||
const ok = await checkRateLimit(
|
||||
`${EVENTS_ACK_LIMIT.scope}:${holderUserId}`,
|
||||
EVENTS_ACK_LIMIT.limit,
|
||||
EVENTS_ACK_LIMIT.window,
|
||||
);
|
||||
if (!ok) throw tooManyCalls();
|
||||
|
||||
const row = await this.stores.durableSubscription.getBySubId(subId);
|
||||
if (
|
||||
!row ||
|
||||
row.holderUserId !== holderUserId ||
|
||||
!rowInActorScope(actor, row)
|
||||
)
|
||||
throw unknownSubscription();
|
||||
|
||||
await this.stores.pendingDelivery.settle(subId, entryId);
|
||||
await this.#drain(row);
|
||||
}
|
||||
|
||||
/**
|
||||
* Drop rows past their expiry, in batches, and report how many went. Every
|
||||
* node sweeps; the delete is idempotent, so two overlapping costs a few
|
||||
@@ -699,6 +835,59 @@ export class EventsService extends PuterService {
|
||||
return removed;
|
||||
}
|
||||
|
||||
/**
|
||||
* Retry what nobody took. Reads the pending index from its oldest end — the
|
||||
* subscriptions that have waited longest — so finding the work costs one
|
||||
* ordered read rather than a walk of the keyspace, which is exactly the
|
||||
* thing that gets expensive when there is a backlog to find.
|
||||
*
|
||||
* A lease that lapsed is what makes a delivery claimable again, so this
|
||||
* needs no notion of failure: it retries whatever is not currently held.
|
||||
*/
|
||||
async sweepPending(): Promise<number> {
|
||||
if (!this.enabled) return 0;
|
||||
|
||||
// A partial write can move a pending set without its share of the
|
||||
// region counter going with it, in either direction. This is what
|
||||
// keeps that drift from being permanent.
|
||||
await this.stores.pendingDelivery
|
||||
.reconcileRegionDepth()
|
||||
.catch((err) => {
|
||||
console.warn('[events] pending counter reconcile failed', err);
|
||||
});
|
||||
|
||||
let attempted = 0;
|
||||
for (const { subId } of await this.stores.pendingDelivery.head(
|
||||
PENDING_SWEEP_SUBSCRIPTIONS,
|
||||
)) {
|
||||
try {
|
||||
const row =
|
||||
await this.stores.durableSubscription.getBySubId(subId);
|
||||
// Nothing left to deliver to, so nothing left to hold.
|
||||
if (!row || this.#isOver(row)) {
|
||||
await this.stores.pendingDelivery.purge(subId);
|
||||
continue;
|
||||
}
|
||||
// A suspended row keeps what it is owed — what happens to that
|
||||
// backlog is the suspension's decision, not the sweeper's. It
|
||||
// goes to the back of the line so it cannot hold the head.
|
||||
if (row.suspendedAt !== null) {
|
||||
await this.stores.pendingDelivery.defer(subId);
|
||||
continue;
|
||||
}
|
||||
attempted += await this.#drain(row);
|
||||
} catch (err) {
|
||||
console.warn('[events] pending sweep failed', subId, err);
|
||||
}
|
||||
}
|
||||
return attempted;
|
||||
}
|
||||
|
||||
/** Past its expiry, so the daily reaper is only a matter of time. */
|
||||
#isOver(row: DurableSubscription): boolean {
|
||||
return row.expiresAt !== null && row.expiresAt <= Date.now() / 1000;
|
||||
}
|
||||
|
||||
/**
|
||||
* Resolve, authorize and compile one subscribe request. Shared so a durable
|
||||
* row cannot be created under a weaker check than a session one.
|
||||
@@ -828,7 +1017,7 @@ export class EventsService extends PuterService {
|
||||
candidates: DispatchSubscription[],
|
||||
actingUserId: number | undefined,
|
||||
): Promise<void> {
|
||||
const rows = candidates.filter(deliverableOverSockets);
|
||||
const rows = candidates.filter(deliverable);
|
||||
if (rows.length === 0) return;
|
||||
|
||||
// One throwaway projection reads the op off the registry entry rather
|
||||
@@ -856,9 +1045,21 @@ export class EventsService extends PuterService {
|
||||
actingUserId === row.holderUserId,
|
||||
seq: seq++,
|
||||
});
|
||||
|
||||
// A `single` is owed rather than sent: it is queued, and never
|
||||
// coalesced or broadcast — collapsing two of them would drop one
|
||||
// the subscription was promised.
|
||||
if (isSingle(row)) {
|
||||
await this.#owe(row, event);
|
||||
continue;
|
||||
}
|
||||
|
||||
const targets = targetsOf(row);
|
||||
this.#coalesce().push(coalesceKey(row.subId, event.subject), {
|
||||
target: deliveryTarget(row),
|
||||
envelope: { subId: row.subId, event },
|
||||
socket: targets.includes('socket'),
|
||||
worker: this.#workerInvocation(row, event),
|
||||
});
|
||||
}
|
||||
|
||||
@@ -951,20 +1152,191 @@ export class EventsService extends PuterService {
|
||||
context: EventContext,
|
||||
reason: GapReason,
|
||||
): void {
|
||||
for (const row of rows)
|
||||
for (const row of rows) {
|
||||
const marker: GapMarker = {
|
||||
id: context.id,
|
||||
subject: subject.subject,
|
||||
op: 'gap',
|
||||
reason,
|
||||
ts: context.ts,
|
||||
};
|
||||
// A marker is a delivery, so it takes the same route its
|
||||
// subscription's events would: queued for a `single`, sent for the
|
||||
// rest.
|
||||
if (isSingle(row)) {
|
||||
void this.#owe(row, marker);
|
||||
continue;
|
||||
}
|
||||
// A marker rides the socket; a row with none has nowhere to hear
|
||||
// it, and sending nothing must not count as a delivery.
|
||||
if (!targetsOf(row).includes('socket')) continue;
|
||||
this.#send({
|
||||
target: deliveryTarget(row),
|
||||
socket: true,
|
||||
envelope: { subId: row.subId, event: marker },
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
// -- Owed deliveries ---------------------------------------------
|
||||
|
||||
/**
|
||||
* Queue a `single` and try to hand it straight over. The queue comes first:
|
||||
* an attempt that fails after it is recorded is a retry, where one that
|
||||
* fails before it is a lost event.
|
||||
*/
|
||||
async #owe(
|
||||
row: DispatchSubscription,
|
||||
event: DeliverableEvent,
|
||||
): Promise<void> {
|
||||
try {
|
||||
const { shed } = await this.stores.pendingDelivery.enqueue(
|
||||
row.subId,
|
||||
event,
|
||||
);
|
||||
this.#reportShed(shed);
|
||||
await this.#drain(row);
|
||||
} catch (err) {
|
||||
this.#enqueueFailed(row, err);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Hand this subscription what it is owed, oldest first, until something
|
||||
* takes a lease and holds it. Stops after a batch so one consumer that
|
||||
* settles inline cannot drain a whole backlog inside one call.
|
||||
*/
|
||||
async #drain(row: DispatchSubscription): Promise<number> {
|
||||
let handed = 0;
|
||||
for (let pass = 0; pass < PENDING_DRAIN_BATCH; pass++) {
|
||||
const claimed = await this.stores.pendingDelivery.claim(row.subId);
|
||||
if (!claimed) return handed;
|
||||
handed++;
|
||||
// Anything still holding the lease is the next consumer's answer to
|
||||
// give, so this pass is over.
|
||||
if (!(await this.#handOut(row, claimed))) return handed;
|
||||
}
|
||||
return handed;
|
||||
}
|
||||
|
||||
/**
|
||||
* One attempt at one owed delivery. Sockets first and one at a time, the
|
||||
* handler once they are spent — a client that is there answers faster than
|
||||
* anything else, and one that is not must not stall the delivery forever.
|
||||
*
|
||||
* Returns whether the delivery settled, which is what says the next one may
|
||||
* go out now rather than when this lease lapses.
|
||||
*/
|
||||
async #handOut(
|
||||
row: DispatchSubscription,
|
||||
claimed: ClaimedDelivery,
|
||||
): Promise<boolean> {
|
||||
const targets = targetsOf(row);
|
||||
const target = deliveryTarget(row);
|
||||
|
||||
// Only this region's own connections are visible here; the ones other
|
||||
// regions hold arrive with presence.
|
||||
const overSocket =
|
||||
targets.includes('socket') &&
|
||||
claimed.socketAttempts < SINGLE_SOCKET_ATTEMPTS &&
|
||||
this.services.socket.has(target);
|
||||
|
||||
if (overSocket) {
|
||||
await this.stores.pendingDelivery.recordSocketAttempt(
|
||||
row.subId,
|
||||
claimed.entryId,
|
||||
);
|
||||
this.#send({
|
||||
target,
|
||||
socket: true,
|
||||
envelope: {
|
||||
subId: row.subId,
|
||||
event: {
|
||||
id: context.id,
|
||||
subject: subject.subject,
|
||||
op: 'gap',
|
||||
reason,
|
||||
ts: context.ts,
|
||||
},
|
||||
event: claimed.event,
|
||||
ackRequired: true,
|
||||
ackId: claimed.entryId,
|
||||
},
|
||||
});
|
||||
return false;
|
||||
}
|
||||
|
||||
// Nowhere to put it yet: the lease is what paces the next attempt.
|
||||
if (!targets.includes('worker')) return false;
|
||||
|
||||
const invocation = this.#workerInvocation(row, claimed.event);
|
||||
if (!invocation) return false;
|
||||
|
||||
const outcome = await this.worker.invoke(invocation);
|
||||
this.onDelivered({ subId: row.subId, event: claimed.event });
|
||||
if (outcome !== 'settled') return false;
|
||||
|
||||
await this.stores.pendingDelivery.settle(row.subId, claimed.entryId);
|
||||
return true;
|
||||
}
|
||||
|
||||
/** What the handler seam is handed, or null for a row that wants none. */
|
||||
#workerInvocation(
|
||||
row: DispatchSubscription,
|
||||
event: DeliverableEvent,
|
||||
): WorkerInvocation | null {
|
||||
if (row.durable !== true) return null;
|
||||
if (!targetsOf(row).includes('worker')) return null;
|
||||
return {
|
||||
subId: row.subId,
|
||||
holderUserId: row.holderUserId,
|
||||
appUid: row.appUid,
|
||||
handlerName: row.handlerName ?? null,
|
||||
event,
|
||||
context: row.context ?? null,
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Say that deliveries were dropped to stay inside a cap. The subscriptions
|
||||
* that lost them are told by the gap marker the store queued in their
|
||||
* place; this is the half nobody else would see.
|
||||
*/
|
||||
#reportShed(shed: readonly PendingShed[]): void {
|
||||
for (const dropped of shed) {
|
||||
if (dropped.scope === 'subscription') {
|
||||
console.warn(
|
||||
`[events] dropped ${dropped.dropped} undelivered event(s) of ${dropped.subId}: backlog full`,
|
||||
);
|
||||
continue;
|
||||
}
|
||||
console.error(
|
||||
`[events] dropped ${dropped.dropped} undelivered event(s) of ${dropped.subId}: too many held in this region`,
|
||||
);
|
||||
this.clients.alarm.create(
|
||||
'events_pending_ceiling',
|
||||
'Too many undelivered events are being held here — the oldest were dropped',
|
||||
{ subId: dropped.subId, dropped: dropped.dropped },
|
||||
'warning',
|
||||
{ dedup: true },
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* A `single` that could not be queued is an event its subscription was
|
||||
* promised and will never see, and nothing downstream will notice on its
|
||||
* own. The write that produced it still succeeded.
|
||||
*/
|
||||
#enqueueFailed(row: DispatchSubscription, err: unknown): void {
|
||||
console.error(
|
||||
`[events] could not queue an event for ${row.subId}`,
|
||||
err,
|
||||
);
|
||||
this.clients.alarm.create(
|
||||
'events_pending_enqueue_failed',
|
||||
'An event owed to a subscription could not be queued and is lost',
|
||||
{
|
||||
subId: row.subId,
|
||||
holderUserId: row.holderUserId,
|
||||
error: err instanceof Error ? err : new Error(String(err)),
|
||||
},
|
||||
'warning',
|
||||
{ dedup: true },
|
||||
);
|
||||
}
|
||||
|
||||
// -- Delivery ----------------------------------------------------
|
||||
@@ -991,6 +1363,7 @@ export class EventsService extends PuterService {
|
||||
const event = delivery.envelope.event as ProjectedEvent;
|
||||
this.#send({
|
||||
target: delivery.target,
|
||||
socket: delivery.socket,
|
||||
envelope: {
|
||||
subId: delivery.envelope.subId,
|
||||
event: {
|
||||
@@ -1010,22 +1383,40 @@ export class EventsService extends PuterService {
|
||||
/**
|
||||
* Addressed at a socket id — which socket.io joins every socket to — or at
|
||||
* a room, so either way the adapter carries it to whichever node terminates
|
||||
* the connection.
|
||||
* the connection. A durable row may also want its handler run, which
|
||||
* happens alongside the socket copy and at most once per delivery.
|
||||
*/
|
||||
#send(delivery: AddressedDelivery): void {
|
||||
try {
|
||||
void this.services.socket
|
||||
.send(
|
||||
delivery.target,
|
||||
EVENTS_DELIVERY_CHANNEL,
|
||||
delivery.envelope,
|
||||
)
|
||||
.catch((err: unknown) => {
|
||||
console.warn('[events] socket send failed', err);
|
||||
});
|
||||
} catch (err) {
|
||||
console.warn('[events] socket send failed', err);
|
||||
if (delivery.socket) {
|
||||
try {
|
||||
void this.services.socket
|
||||
.send(
|
||||
delivery.target,
|
||||
EVENTS_DELIVERY_CHANNEL,
|
||||
delivery.envelope,
|
||||
)
|
||||
.catch((err: unknown) => {
|
||||
console.warn('[events] socket send failed', err);
|
||||
});
|
||||
} catch (err) {
|
||||
console.warn('[events] socket send failed', err);
|
||||
}
|
||||
}
|
||||
|
||||
if (delivery.worker) {
|
||||
const invocation = delivery.worker;
|
||||
// At-most-once by construction: a `broadcast` invocation is never
|
||||
// retried, which is why the docs ask handlers to be idempotent
|
||||
// rather than promising them each event exactly once.
|
||||
try {
|
||||
void this.worker.invoke(invocation).catch((err: unknown) => {
|
||||
console.warn('[events] handler invocation failed', err);
|
||||
});
|
||||
} catch (err) {
|
||||
console.warn('[events] handler invocation failed', err);
|
||||
}
|
||||
}
|
||||
|
||||
this.onDelivered(delivery.envelope);
|
||||
}
|
||||
|
||||
@@ -1205,6 +1596,17 @@ export class EventsService extends PuterService {
|
||||
this.#expirySweep = sweep;
|
||||
}
|
||||
|
||||
#armPendingSweep(): void {
|
||||
if (!this.enabled) return;
|
||||
const sweep = setInterval(() => {
|
||||
void this.sweepPending().catch((err) => {
|
||||
console.warn('[events] pending sweep failed', err);
|
||||
});
|
||||
}, PENDING_SWEEP_INTERVAL_MS);
|
||||
sweep.unref?.();
|
||||
this.#pendingSweep = sweep;
|
||||
}
|
||||
|
||||
#stopRefresh(holderUserId: number, socketId: string): void {
|
||||
const key = `${holderUserId}|${socketId}`;
|
||||
const timer = this.#refreshTimers.get(key);
|
||||
|
||||
@@ -220,14 +220,39 @@ describe('creating a durable subscription over HTTP', () => {
|
||||
expect(created.body.context).toBeUndefined();
|
||||
});
|
||||
|
||||
it('refuses the delivery class that has nowhere to queue yet', async () => {
|
||||
const refused = await subscribe(env.users.user.token, {
|
||||
it('registers a subscription owed to one consumer', async () => {
|
||||
await clearRows();
|
||||
const created = await subscribe(env.users.user.token, {
|
||||
delivery: 'single',
|
||||
handlerName: 'onWrite',
|
||||
});
|
||||
|
||||
expect(refused.status).toBe(501);
|
||||
expect(refused.body.code).toBe('delivery_class_unavailable');
|
||||
expect(created.status).toBe(200);
|
||||
expect(created.body).toMatchObject({
|
||||
delivery: 'single',
|
||||
handlerName: 'onWrite',
|
||||
targets: ['socket', 'worker'],
|
||||
});
|
||||
});
|
||||
|
||||
it('refuses one owed to a consumer it cannot name', async () => {
|
||||
const refused = await subscribe(env.users.user.token, {
|
||||
delivery: 'single',
|
||||
});
|
||||
|
||||
expect(refused.status).toBe(400);
|
||||
expect(refused.body.code).toBe('events_handler_required');
|
||||
});
|
||||
|
||||
it('refuses one owed to a device notification', async () => {
|
||||
const refused = await subscribe(env.users.user.token, {
|
||||
delivery: 'single',
|
||||
handlerName: 'onWrite',
|
||||
targets: ['push'],
|
||||
});
|
||||
|
||||
expect(refused.status).toBe(400);
|
||||
expect(refused.body.code).toBe('invalid_targets');
|
||||
});
|
||||
|
||||
it('refuses a target outside the known set', async () => {
|
||||
@@ -239,6 +264,17 @@ describe('creating a durable subscription over HTTP', () => {
|
||||
expect(refused.body.code).toBe('invalid_targets');
|
||||
});
|
||||
|
||||
it('refuses a `single` subscription with no handler to fall back to', async () => {
|
||||
const refused = await subscribe(env.users.user.token, {
|
||||
delivery: 'single',
|
||||
handlerName: 'onWrite',
|
||||
targets: ['socket'],
|
||||
});
|
||||
|
||||
expect(refused.status).toBe(400);
|
||||
expect(refused.body.code).toBe('invalid_targets');
|
||||
});
|
||||
|
||||
it('refuses an expiry in the past', async () => {
|
||||
const refused = await subscribe(env.users.user.token, {
|
||||
expiresAt: Math.floor(Date.now() / 1000) - 60,
|
||||
|
||||
@@ -46,6 +46,29 @@ export interface ProjectedEvent {
|
||||
seq: number;
|
||||
}
|
||||
|
||||
export type GapReason =
|
||||
| 'matched_subscription_limit'
|
||||
| 'filter_evaluation_limit'
|
||||
| 'delivery_rate_limit'
|
||||
| 'backlog_overflow';
|
||||
|
||||
/**
|
||||
* `gap` says an event existed and was not delivered. It rides the delivery
|
||||
* channel because a subscriber that never saw one would read the silence as
|
||||
* "nothing happened", and carries no `uid`/`path` — what was dropped is exactly
|
||||
* what it cannot name.
|
||||
*/
|
||||
export interface GapMarker {
|
||||
id: string;
|
||||
subject: string;
|
||||
op: 'gap';
|
||||
reason: GapReason;
|
||||
ts: number;
|
||||
}
|
||||
|
||||
/** Either shape a subscriber can be handed. */
|
||||
export type DeliverableEvent = ProjectedEvent | GapMarker;
|
||||
|
||||
/** What dispatch knows about one internal emit, before it picks subscribers. */
|
||||
export interface EventContext {
|
||||
key: EventKey;
|
||||
|
||||
@@ -0,0 +1,499 @@
|
||||
/*
|
||||
* Copyright (C) 2024-present Puter Technologies Inc.
|
||||
*
|
||||
* This file is part of Puter.
|
||||
*
|
||||
* Puter is free software: you can redistribute it and/or modify
|
||||
* it under the terms of the GNU Affero General Public License as published
|
||||
* by the Free Software Foundation, either version 3 of the License, or
|
||||
* (at your option) any later version.
|
||||
*
|
||||
* This program is distributed in the hope that it will be useful,
|
||||
* but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
* GNU Affero General Public License for more details.
|
||||
*
|
||||
* You should have received a copy of the GNU Affero General Public License
|
||||
* along with this program. If not, see <https://www.gnu.org/licenses/>.
|
||||
*/
|
||||
|
||||
import MockRedis from 'ioredis-mock';
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
|
||||
import {
|
||||
EVENTS_COALESCE_WINDOW_MS,
|
||||
EVENTS_REGION_PENDING_CEILING,
|
||||
} from '../../controllers/events/limits.js';
|
||||
import type { Actor } from '../../core/actor.js';
|
||||
import { EventSubscriptionStore } from '../../stores/events/EventSubscriptionStore.js';
|
||||
import { PendingDeliveryStore } from '../../stores/events/PendingDeliveryStore.js';
|
||||
import type {
|
||||
DurableSubscription,
|
||||
SubscriptionTarget,
|
||||
} from '../../stores/events/types.js';
|
||||
import type { FSEntry } from '../../stores/fs/FSEntry.js';
|
||||
import type { IConfig } from '../../types.js';
|
||||
import { EventsService, type DeliveryEnvelope } from './EventsService.js';
|
||||
import { fsAnchorToken } from './subjects.js';
|
||||
import type {
|
||||
WorkerInvocation,
|
||||
WorkerInvocationOutcome,
|
||||
} from './workerSeam.js';
|
||||
|
||||
/**
|
||||
* The promises each delivery class makes, held against each other.
|
||||
*
|
||||
* A `single` is owed to exactly one consumer: the socket if someone is there,
|
||||
* the handler once the socket has had its turns, and never both. A `broadcast`
|
||||
* is at-most-once to everyone who is: its handler runs alongside the socket
|
||||
* copies rather than instead of them, and a row nothing can carry delivers —
|
||||
* and meters — nothing at all.
|
||||
*/
|
||||
|
||||
let seq = 0;
|
||||
let userId = 0;
|
||||
let redis: InstanceType<typeof MockRedis.Cluster>;
|
||||
let subscriptions: EventSubscriptionStore;
|
||||
let pending: PendingDeliveryStore;
|
||||
let service: EventsService;
|
||||
let sent: DeliveryEnvelope[];
|
||||
let delivered: DeliveryEnvelope[];
|
||||
let invoked: WorkerInvocation[];
|
||||
let rows: Map<string, DurableSubscription>;
|
||||
let entries: Map<string, FSEntry>;
|
||||
let alarms: ReturnType<typeof vi.fn>;
|
||||
|
||||
/** Whether this region holds a connection for the row being delivered to. */
|
||||
let socketConnected = true;
|
||||
/** What the handler seam reports back, which is what settles a lease or not. */
|
||||
let workerOutcome: WorkerInvocationOutcome = 'deferred';
|
||||
|
||||
const entry = (over: Partial<FSEntry> = {}): FSEntry =>
|
||||
({
|
||||
uid: `file-${seq}`,
|
||||
uuid: `file-${seq}`,
|
||||
path: `/u${userId}/Documents/notes.txt`,
|
||||
userId,
|
||||
isDir: false,
|
||||
...over,
|
||||
}) as FSEntry;
|
||||
|
||||
const actorFor = (asUserId = userId): Actor =>
|
||||
({
|
||||
user: {
|
||||
id: asUserId,
|
||||
uuid: `user-${asUserId}`,
|
||||
username: `u${asUserId}`,
|
||||
},
|
||||
effectiveApp: null,
|
||||
}) as unknown as Actor;
|
||||
|
||||
const anchorUid = (): string => `docs-${seq}`;
|
||||
const anchorPath = (): string => `/u${userId}/Documents`;
|
||||
|
||||
const ancestors = (): Array<{ uid: string; path: string }> => [
|
||||
{ uid: anchorUid(), path: anchorPath() },
|
||||
];
|
||||
|
||||
const durableRow = (
|
||||
over: Partial<DurableSubscription> = {},
|
||||
): DurableSubscription => ({
|
||||
durable: true,
|
||||
subId: `app-${seq}#${over.subId ?? 'sub'}`,
|
||||
holderUserId: userId,
|
||||
ownerUserId: userId,
|
||||
subject: `fs:${anchorPath()}`,
|
||||
token: fsAnchorToken(anchorUid()),
|
||||
anchorUid: anchorUid(),
|
||||
anchorPath: anchorPath(),
|
||||
match: null,
|
||||
op: null,
|
||||
appUid: null,
|
||||
permission: 'list',
|
||||
delivery: 'single',
|
||||
targets: ['socket', 'worker'] as SubscriptionTarget[],
|
||||
handlerName: 'onWrite',
|
||||
context: null,
|
||||
expiresAt: null,
|
||||
suspendedAt: null,
|
||||
suspendedReason: null,
|
||||
createdAt: Math.floor(Date.now() / 1000),
|
||||
...over,
|
||||
});
|
||||
|
||||
/** Put a row where dispatch reads it, and where a later ack looks it up. */
|
||||
const register = async (
|
||||
over: Partial<DurableSubscription> = {},
|
||||
): Promise<DurableSubscription> => {
|
||||
const row = durableRow(over);
|
||||
rows.set(row.subId, row);
|
||||
await subscriptions.cacheDurable([row]);
|
||||
service.invalidateUser(userId);
|
||||
return row;
|
||||
};
|
||||
|
||||
const dispatch = (node = entry()): Promise<void> =>
|
||||
service.dispatchFs('fs.write.file', node, {
|
||||
actingUserId: userId,
|
||||
ancestors: async () => ancestors(),
|
||||
});
|
||||
|
||||
/** Wait out the coalescing window a `broadcast` delivery sits in. */
|
||||
const flushed = (count = 1): Promise<void> =>
|
||||
vi.waitFor(() => expect(sent.length).toBeGreaterThanOrEqual(count), {
|
||||
timeout: EVENTS_COALESCE_WINDOW_MS * 12,
|
||||
interval: 25,
|
||||
});
|
||||
|
||||
const jump = (ms: number): void => {
|
||||
vi.useFakeTimers({ toFake: ['Date'] });
|
||||
vi.setSystemTime(Date.now() + ms);
|
||||
};
|
||||
|
||||
const keysOf = async (subId: string): Promise<string[]> =>
|
||||
(await redis.keys('*')).filter((key: string) => key.includes(subId));
|
||||
|
||||
beforeEach(async () => {
|
||||
seq++;
|
||||
userId = 7000 + seq;
|
||||
socketConnected = true;
|
||||
workerOutcome = 'deferred';
|
||||
sent = [];
|
||||
delivered = [];
|
||||
invoked = [];
|
||||
rows = new Map();
|
||||
entries = new Map();
|
||||
alarms = vi.fn();
|
||||
|
||||
redis = new MockRedis.Cluster(['redis://localhost:7001']);
|
||||
await redis.del('ev:qx', 'ev:qc');
|
||||
|
||||
subscriptions = new EventSubscriptionStore(
|
||||
{} as IConfig,
|
||||
{ redis } as never,
|
||||
{} as never,
|
||||
);
|
||||
pending = new PendingDeliveryStore(
|
||||
{} as IConfig,
|
||||
{ redis } as never,
|
||||
{} as never,
|
||||
);
|
||||
|
||||
entries.set(`uid:${anchorUid()}`, entry({ uid: anchorUid(), isDir: true }));
|
||||
|
||||
service = new EventsService(
|
||||
{ events: { enabled: true } } as IConfig,
|
||||
{
|
||||
redis,
|
||||
event: { on: vi.fn(), emit: vi.fn() },
|
||||
alarm: { create: alarms },
|
||||
} as never,
|
||||
{
|
||||
eventSubscription: subscriptions,
|
||||
pendingDelivery: pending,
|
||||
durableSubscription: {
|
||||
warmRegion: async () => false,
|
||||
getBySubId: async (subId: string) => rows.get(subId) ?? null,
|
||||
remove: async (row: DurableSubscription) => {
|
||||
rows.delete(row.subId);
|
||||
return { userId: row.holderUserId, generation: 1 };
|
||||
},
|
||||
},
|
||||
fsEntry: {
|
||||
getEntryByUuid: async (uid: string) =>
|
||||
entries.get(`uid:${uid}`) ?? null,
|
||||
getEntryByPath: async () => null,
|
||||
getEntryById: async () => null,
|
||||
},
|
||||
user: {
|
||||
getById: async (id: number) => ({ id, uuid: `user-${id}` }),
|
||||
},
|
||||
app: { getByUid: async (uid: string) => ({ uid, id: 1 }) },
|
||||
} as never,
|
||||
{
|
||||
socket: {
|
||||
send: vi.fn(async (_spec, _key, data) => {
|
||||
sent.push(data as DeliveryEnvelope);
|
||||
}),
|
||||
has: () => socketConnected,
|
||||
},
|
||||
fs: { getAncestorChain: async () => ancestors() },
|
||||
acl: {
|
||||
check: async () => true,
|
||||
getSafeAclError: async () => ({
|
||||
status: 404,
|
||||
message: 'Subject does not exist',
|
||||
fields: { code: 'subject_does_not_exist' },
|
||||
}),
|
||||
},
|
||||
} as never,
|
||||
);
|
||||
service.onDelivered = (envelope) => delivered.push(envelope);
|
||||
service.worker = {
|
||||
invoke: async (invocation: WorkerInvocation) => {
|
||||
invoked.push(invocation);
|
||||
return workerOutcome;
|
||||
},
|
||||
};
|
||||
});
|
||||
|
||||
afterEach(() => {
|
||||
vi.useRealTimers();
|
||||
});
|
||||
|
||||
describe('a delivery owed to exactly one consumer', () => {
|
||||
it('goes to the connected socket, and not to the handler as well', async () => {
|
||||
const row = await register();
|
||||
|
||||
await dispatch();
|
||||
|
||||
expect(invoked).toEqual([]);
|
||||
expect(sent).toHaveLength(1);
|
||||
expect(sent[0]).toMatchObject({
|
||||
subId: row.subId,
|
||||
ackRequired: true,
|
||||
event: { op: 'write' },
|
||||
});
|
||||
expect(sent[0].ackId).toEqual(expect.any(String));
|
||||
expect(delivered).toHaveLength(1);
|
||||
});
|
||||
|
||||
it('stays owed until the client acks it', async () => {
|
||||
const row = await register();
|
||||
await dispatch();
|
||||
|
||||
await expect(pending.depth(row.subId)).resolves.toBe(1);
|
||||
|
||||
await service.ackDelivery(actorFor(), {
|
||||
subId: row.subId,
|
||||
id: sent[0].ackId,
|
||||
});
|
||||
|
||||
await expect(pending.depth(row.subId)).resolves.toBe(0);
|
||||
// Nothing owed, nothing held.
|
||||
await expect(keysOf(row.subId)).resolves.toEqual([]);
|
||||
});
|
||||
|
||||
it('takes a second ack for the same delivery as nothing to do', async () => {
|
||||
const row = await register();
|
||||
await dispatch();
|
||||
const ack = { subId: row.subId, id: sent[0].ackId };
|
||||
|
||||
await service.ackDelivery(actorFor(), ack);
|
||||
await expect(
|
||||
service.ackDelivery(actorFor(), ack),
|
||||
).resolves.toBeUndefined();
|
||||
});
|
||||
|
||||
it('refuses to settle a delivery for a subscription someone else holds', async () => {
|
||||
const row = await register();
|
||||
await dispatch();
|
||||
const ack = { subId: row.subId, id: sent[0].ackId };
|
||||
|
||||
await expect(
|
||||
service.ackDelivery(actorFor(userId + 1), ack),
|
||||
).rejects.toMatchObject({
|
||||
statusCode: 404,
|
||||
legacyCode: 'subscription_does_not_exist',
|
||||
});
|
||||
|
||||
// Untouched: the real holder can still take it.
|
||||
await expect(pending.depth(row.subId)).resolves.toBe(1);
|
||||
await service.ackDelivery(actorFor(), ack);
|
||||
await expect(pending.depth(row.subId)).resolves.toBe(0);
|
||||
});
|
||||
|
||||
it('hands the next one over as soon as the last is settled', async () => {
|
||||
const row = await register();
|
||||
await dispatch(entry({ uid: `file-a-${seq}` }));
|
||||
await dispatch(entry({ uid: `file-b-${seq}` }));
|
||||
|
||||
// One at a time: the second waits on the first being taken.
|
||||
expect(sent).toHaveLength(1);
|
||||
|
||||
await service.ackDelivery(actorFor(), {
|
||||
subId: row.subId,
|
||||
id: sent[0].ackId,
|
||||
});
|
||||
|
||||
expect(sent).toHaveLength(2);
|
||||
expect(sent[1].event.id).not.toBe(sent[0].event.id);
|
||||
});
|
||||
|
||||
it('tries the socket twice, then hands it to the handler', async () => {
|
||||
const row = await register();
|
||||
await dispatch();
|
||||
expect(sent).toHaveLength(1);
|
||||
|
||||
// First lease lapses with no ack: a second socket attempt, which is
|
||||
// what a reconnected client is worth.
|
||||
jump(31_000);
|
||||
await service.sweepPending();
|
||||
expect(sent).toHaveLength(2);
|
||||
expect(invoked).toEqual([]);
|
||||
|
||||
// Second lapses too. Sockets are spent, so the handler takes it.
|
||||
jump(31_000);
|
||||
await service.sweepPending();
|
||||
expect(sent).toHaveLength(2);
|
||||
expect(invoked).toHaveLength(1);
|
||||
expect(invoked[0]).toMatchObject({
|
||||
subId: row.subId,
|
||||
handlerName: 'onWrite',
|
||||
holderUserId: userId,
|
||||
});
|
||||
});
|
||||
|
||||
it('goes straight to the handler when nothing is connected', async () => {
|
||||
socketConnected = false;
|
||||
const row = await register();
|
||||
|
||||
await dispatch();
|
||||
|
||||
expect(sent).toEqual([]);
|
||||
expect(invoked).toHaveLength(1);
|
||||
// Not taken yet, so still owed.
|
||||
await expect(pending.depth(row.subId)).resolves.toBe(1);
|
||||
});
|
||||
|
||||
it('settles the delivery once the handler reports it took it', async () => {
|
||||
socketConnected = false;
|
||||
workerOutcome = 'settled';
|
||||
const row = await register();
|
||||
|
||||
await dispatch();
|
||||
|
||||
expect(invoked).toHaveLength(1);
|
||||
await expect(pending.depth(row.subId)).resolves.toBe(0);
|
||||
await expect(keysOf(row.subId)).resolves.toEqual([]);
|
||||
});
|
||||
|
||||
it('is never delivered as a broadcast copy as well', async () => {
|
||||
await register();
|
||||
|
||||
await dispatch();
|
||||
await new Promise((resolve) =>
|
||||
setTimeout(resolve, EVENTS_COALESCE_WINDOW_MS * 3),
|
||||
);
|
||||
|
||||
expect(sent).toHaveLength(1);
|
||||
expect(sent[0].ackRequired).toBe(true);
|
||||
});
|
||||
});
|
||||
|
||||
describe('a delivery everyone connected gets', () => {
|
||||
it('runs the handler once, alongside the socket copy', async () => {
|
||||
const row = await register({ delivery: 'broadcast' });
|
||||
|
||||
await dispatch();
|
||||
await flushed();
|
||||
|
||||
expect(sent).toHaveLength(1);
|
||||
expect(sent[0]).toMatchObject({ subId: row.subId });
|
||||
expect(sent[0].ackRequired).toBeUndefined();
|
||||
expect(invoked).toHaveLength(1);
|
||||
expect(invoked[0]).toMatchObject({ subId: row.subId });
|
||||
// One delivery is one line, however many transports carried it.
|
||||
expect(delivered).toHaveLength(1);
|
||||
});
|
||||
|
||||
it('runs the handler once per delivery, not once per transport', async () => {
|
||||
await register({ delivery: 'broadcast', targets: ['worker'] });
|
||||
|
||||
await dispatch();
|
||||
await vi.waitFor(() => expect(invoked.length).toBe(1), {
|
||||
timeout: EVENTS_COALESCE_WINDOW_MS * 12,
|
||||
interval: 25,
|
||||
});
|
||||
|
||||
expect(sent).toEqual([]);
|
||||
expect(delivered).toHaveLength(1);
|
||||
});
|
||||
|
||||
it('delivers and meters nothing when nothing can carry it', async () => {
|
||||
await register({ delivery: 'broadcast', targets: ['push'] });
|
||||
|
||||
await dispatch();
|
||||
await new Promise((resolve) =>
|
||||
setTimeout(resolve, EVENTS_COALESCE_WINDOW_MS * 3),
|
||||
);
|
||||
|
||||
expect(sent).toEqual([]);
|
||||
expect(invoked).toEqual([]);
|
||||
expect(delivered).toEqual([]);
|
||||
});
|
||||
});
|
||||
|
||||
/** No alarm text may name the storage this runs on — see AGENTS.md. */
|
||||
const noInfraWording = (): RegExp =>
|
||||
/^(?!.*\b(redis|elasticache|dynamo|dynamodb|aws|memcached)\b).*$/is;
|
||||
|
||||
describe('when a delivery cannot be held', () => {
|
||||
it('does not fail the write, and does not let the loss pass quietly', async () => {
|
||||
await register();
|
||||
vi.spyOn(pending, 'enqueue').mockRejectedValue(
|
||||
new Error('the cache is unreachable'),
|
||||
);
|
||||
|
||||
await expect(dispatch()).resolves.toBeUndefined();
|
||||
|
||||
expect(sent).toEqual([]);
|
||||
expect(alarms).toHaveBeenCalledWith(
|
||||
'events_pending_enqueue_failed',
|
||||
expect.stringMatching(noInfraWording()),
|
||||
expect.objectContaining({ subId: expect.any(String) }),
|
||||
'warning',
|
||||
{ dedup: true },
|
||||
);
|
||||
});
|
||||
|
||||
it('sheds the oldest and alarms when the region is holding too much', async () => {
|
||||
const row = await register();
|
||||
await redis.incrby('ev:qc', EVENTS_REGION_PENDING_CEILING);
|
||||
|
||||
await dispatch();
|
||||
|
||||
expect(alarms).toHaveBeenCalledWith(
|
||||
'events_pending_ceiling',
|
||||
expect.stringMatching(noInfraWording()),
|
||||
expect.objectContaining({ subId: row.subId }),
|
||||
'warning',
|
||||
{ dedup: true },
|
||||
);
|
||||
// What it shed, it was told about: the marker took the delivery's place.
|
||||
expect(sent).toHaveLength(1);
|
||||
expect(sent[0].event).toMatchObject({
|
||||
op: 'gap',
|
||||
reason: 'backlog_overflow',
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
describe('giving up a subscription', () => {
|
||||
it('drops what it was owed, and corrects the region counter', async () => {
|
||||
socketConnected = false;
|
||||
const row = await register();
|
||||
await dispatch();
|
||||
await expect(pending.regionDepth()).resolves.toBe(1);
|
||||
|
||||
await service.unsubscribeDurable(actorFor(), { subId: row.subId });
|
||||
|
||||
await expect(pending.depth(row.subId)).resolves.toBe(0);
|
||||
await expect(pending.regionDepth()).resolves.toBe(0);
|
||||
await expect(keysOf(row.subId)).resolves.toEqual([]);
|
||||
});
|
||||
});
|
||||
|
||||
describe('keeping the region counter honest', () => {
|
||||
it('self-heals a drifted counter on the periodic sweep', async () => {
|
||||
socketConnected = false;
|
||||
await register();
|
||||
await dispatch();
|
||||
// Drift: as if a settle's decrement had been lost somewhere else.
|
||||
await redis.incrby('ev:qc', 41);
|
||||
await expect(pending.regionDepth()).resolves.toBe(42);
|
||||
|
||||
await service.sweepPending();
|
||||
|
||||
await expect(pending.regionDepth()).resolves.toBe(1);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,76 @@
|
||||
/*
|
||||
* Copyright (C) 2024-present Puter Technologies Inc.
|
||||
*
|
||||
* This file is part of Puter.
|
||||
*
|
||||
* Puter is free software: you can redistribute it and/or modify
|
||||
* it under the terms of the GNU Affero General Public License as published
|
||||
* by the Free Software Foundation, either version 3 of the License, or
|
||||
* (at your option) any later version.
|
||||
*
|
||||
* This program is distributed in the hope that it will be useful,
|
||||
* but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
* GNU Affero General Public License for more details.
|
||||
*
|
||||
* You should have received a copy of the GNU Affero General Public License
|
||||
* along with this program. If not, see <https://www.gnu.org/licenses/>.
|
||||
*/
|
||||
|
||||
import type { DeliverableEvent } from './registry.js';
|
||||
|
||||
/**
|
||||
* Where a delivery leaves the event system for the app's own code.
|
||||
*
|
||||
* The invoker itself — minting the subscriber-scoped token, the call, its
|
||||
* retries — is not built yet, and delivery semantics must not wait for it: what
|
||||
* runs a handler is one decision, and how many times a `single` may be handed
|
||||
* out is another. So the seam is the boundary, and the default records the
|
||||
* intent without acting on it.
|
||||
*
|
||||
* An invocation is not an ack. A `single` delivery stays leased until the
|
||||
* invoker reports the handler took it, which is what keeps a handler that never
|
||||
* ran from looking like one that succeeded.
|
||||
*/
|
||||
|
||||
/** One handler call, as the seam receives it. */
|
||||
export interface WorkerInvocation {
|
||||
subId: string;
|
||||
/** Whose subscription this is, and who is billed for the work. */
|
||||
holderUserId: number;
|
||||
/** The app whose handler runs, or null for an account-owned row. */
|
||||
appUid: string | null;
|
||||
handlerName: string | null;
|
||||
event: DeliverableEvent;
|
||||
/** The subscription's stored context, delivered to the handler as `ctx`. */
|
||||
context: string | null;
|
||||
}
|
||||
|
||||
/**
|
||||
* What the invoker did with it. `settled` means the handler took the delivery
|
||||
* and its lease may be released; `deferred` means it has not, and the delivery
|
||||
* stays owed.
|
||||
*/
|
||||
export type WorkerInvocationOutcome = 'settled' | 'deferred';
|
||||
|
||||
export interface WorkerInvokerSeam {
|
||||
invoke(invocation: WorkerInvocation): Promise<WorkerInvocationOutcome>;
|
||||
}
|
||||
|
||||
/** Invocations one recorder holds, so it cannot grow with event volume. */
|
||||
const RECORDED_INVOCATIONS = 100;
|
||||
|
||||
/**
|
||||
* The seam until there is something to run handlers. Records what would have
|
||||
* been invoked and settles nothing, so a delivery handed here stays owed and is
|
||||
* retried rather than quietly disappearing.
|
||||
*/
|
||||
export class RecordingWorkerInvoker implements WorkerInvokerSeam {
|
||||
readonly recorded: WorkerInvocation[] = [];
|
||||
|
||||
invoke(invocation: WorkerInvocation): Promise<WorkerInvocationOutcome> {
|
||||
this.recorded.push(invocation);
|
||||
if (this.recorded.length > RECORDED_INVOCATIONS) this.recorded.shift();
|
||||
return Promise.resolve('deferred');
|
||||
}
|
||||
}
|
||||
@@ -173,6 +173,38 @@ describe('validation at the row write', () => {
|
||||
).rejects.toSatisfy(codeOf('invalid_targets'));
|
||||
});
|
||||
|
||||
it('refuses a `single` row that wants a device notification', async () => {
|
||||
await expect(
|
||||
durable().create(
|
||||
input({
|
||||
delivery: 'single',
|
||||
handlerName: 'onWrite',
|
||||
targets: ['socket', 'push'],
|
||||
}),
|
||||
),
|
||||
).rejects.toSatisfy(codeOf('invalid_targets'));
|
||||
await expect(durable().countForHolder(userId)).resolves.toBe(0);
|
||||
});
|
||||
|
||||
it('refuses a `single` row with no handler to fall back to', async () => {
|
||||
await expect(
|
||||
durable().create(
|
||||
input({
|
||||
delivery: 'single',
|
||||
handlerName: 'onWrite',
|
||||
targets: ['socket'],
|
||||
}),
|
||||
),
|
||||
).rejects.toSatisfy(codeOf('invalid_targets'));
|
||||
});
|
||||
|
||||
it('keeps the same targets on a `broadcast` row, where push is fine', async () => {
|
||||
const { row } = await durable().create(
|
||||
input({ targets: ['socket', 'push'] }),
|
||||
);
|
||||
expect(row.targets).toEqual(['socket', 'push']);
|
||||
});
|
||||
|
||||
it('refuses a context past the hard cap', async () => {
|
||||
await expect(
|
||||
durable().create(input({ context: 'x'.repeat(4097) })),
|
||||
|
||||
@@ -33,6 +33,7 @@ import type { GenerationBump } from './EventSubscriptionStore.js';
|
||||
import {
|
||||
isSubscriptionTarget,
|
||||
SUBSCRIPTION_TARGETS,
|
||||
targetsAllowedForDelivery,
|
||||
type DurableSubscription,
|
||||
type SubscriptionTarget,
|
||||
} from './types.js';
|
||||
@@ -111,6 +112,13 @@ const invalidTargets = (): HttpError =>
|
||||
{ legacyCode: 'invalid_targets' },
|
||||
);
|
||||
|
||||
const pushOnSingle = (): HttpError =>
|
||||
new HttpError(
|
||||
400,
|
||||
'A `single` subscription needs a `worker` target and may not target `push`',
|
||||
{ legacyCode: 'invalid_targets' },
|
||||
);
|
||||
|
||||
const quotaReached = (): HttpError =>
|
||||
new HttpError(
|
||||
429,
|
||||
@@ -202,7 +210,7 @@ export class DurableSubscriptionStore extends PuterStore {
|
||||
async create(
|
||||
input: DurableSubscriptionInput,
|
||||
): Promise<{ row: DurableSubscription; bump: GenerationBump }> {
|
||||
const targets = this.#assertTargets(input.targets);
|
||||
const targets = this.#assertTargets(input.delivery, input.targets);
|
||||
this.#assertContext(input.context);
|
||||
|
||||
const held = await this.countForHolder(input.holderUserId);
|
||||
@@ -437,11 +445,22 @@ export class DurableSubscriptionStore extends PuterStore {
|
||||
return rows.map(toRow);
|
||||
}
|
||||
|
||||
#assertTargets(targets: readonly string[]): SubscriptionTarget[] {
|
||||
/**
|
||||
* The row cannot exist with transports its delivery class cannot use. Held
|
||||
* here rather than only at the API, so a writer that never passes through
|
||||
* one cannot leave an unsatisfiable row behind.
|
||||
*/
|
||||
#assertTargets(
|
||||
delivery: DeliveryClass,
|
||||
targets: readonly string[],
|
||||
): SubscriptionTarget[] {
|
||||
if (!Array.isArray(targets) || targets.length === 0)
|
||||
throw invalidTargets();
|
||||
if (!targets.every(isSubscriptionTarget)) throw invalidTargets();
|
||||
return [...new Set(targets as SubscriptionTarget[])];
|
||||
|
||||
const unique = [...new Set(targets as SubscriptionTarget[])];
|
||||
if (!targetsAllowedForDelivery(delivery, unique)) throw pushOnSingle();
|
||||
return unique;
|
||||
}
|
||||
|
||||
#assertContext(context: string | null): void {
|
||||
|
||||
@@ -0,0 +1,427 @@
|
||||
/*
|
||||
* Copyright (C) 2024-present Puter Technologies Inc.
|
||||
*
|
||||
* This file is part of Puter.
|
||||
*
|
||||
* Puter is free software: you can redistribute it and/or modify
|
||||
* it under the terms of the GNU Affero General Public License as published
|
||||
* by the Free Software Foundation, either version 3 of the License, or
|
||||
* (at your option) any later version.
|
||||
*
|
||||
* This program is distributed in the hope that it will be useful,
|
||||
* but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
* GNU Affero General Public License for more details.
|
||||
*
|
||||
* You should have received a copy of the GNU Affero General Public License
|
||||
* along with this program. If not, see <https://www.gnu.org/licenses/>.
|
||||
*/
|
||||
|
||||
import MockRedis from 'ioredis-mock';
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
|
||||
import {
|
||||
EVENTS_PENDING_DELIVERIES_PER_SUBSCRIPTION,
|
||||
EVENTS_REGION_PENDING_CEILING,
|
||||
} from '../../controllers/events/limits.js';
|
||||
import type { ProjectedEvent } from '../../services/events/registry.js';
|
||||
import type { IConfig } from '../../types.js';
|
||||
import { PendingDeliveryStore } from './PendingDeliveryStore.js';
|
||||
|
||||
/**
|
||||
* What a subscription is owed, and what holding it costs. Two claims run
|
||||
* through everything here: a delivery is never lost quietly — a drop is a gap
|
||||
* marker the subscriber receives — and a subscription that is keeping up owns
|
||||
* nothing at all.
|
||||
*/
|
||||
|
||||
// The keyspace is shared per process, so each test gets its own subscription
|
||||
// and the two region-wide keys are cleared between them.
|
||||
let seq = 0;
|
||||
let subId = '';
|
||||
let redis: InstanceType<typeof MockRedis.Cluster>;
|
||||
let store: PendingDeliveryStore;
|
||||
let commands: string[];
|
||||
|
||||
const INDEX_KEY = 'ev:qx';
|
||||
const COUNTER_KEY = 'ev:qc';
|
||||
|
||||
const entriesKey = (id = subId): string => `ev:q:{${id}}`;
|
||||
const pendingKey = (id = subId): string => `ev:qp:{${id}}`;
|
||||
|
||||
/** Every command that crossed the client, so a test can say what was not. */
|
||||
const recordingRedis = (
|
||||
inner: InstanceType<typeof MockRedis.Cluster>,
|
||||
): InstanceType<typeof MockRedis.Cluster> =>
|
||||
new Proxy(inner, {
|
||||
get(target, property, receiver) {
|
||||
const value = Reflect.get(target, property, receiver);
|
||||
if (typeof property !== 'string' || typeof value !== 'function')
|
||||
return value;
|
||||
return (...args: unknown[]) => {
|
||||
commands.push(property);
|
||||
return (value as (...a: unknown[]) => unknown).apply(
|
||||
target,
|
||||
args,
|
||||
);
|
||||
};
|
||||
},
|
||||
}) as InstanceType<typeof MockRedis.Cluster>;
|
||||
|
||||
const event = (id: string): ProjectedEvent => ({
|
||||
id,
|
||||
subject: 'fs:/u/Documents',
|
||||
op: 'write',
|
||||
uid: `uid-${id}`,
|
||||
path: `/u/Documents/${id}.txt`,
|
||||
self: true,
|
||||
ts: Date.now(),
|
||||
seq: 0,
|
||||
});
|
||||
|
||||
/** Keys this subscription owns right now, whatever their type. */
|
||||
const keysOf = async (id = subId): Promise<string[]> =>
|
||||
(await redis.keys('*')).filter((key: string) => key.includes(id));
|
||||
|
||||
/**
|
||||
* Put a backlog in place without paying for it one delivery at a time. The caps
|
||||
* are the point of these tests, not the path that fills them.
|
||||
*/
|
||||
const seedBacklog = async (id: string, count: number): Promise<void> => {
|
||||
const scored: Array<string | number> = [];
|
||||
const fields: string[] = [];
|
||||
const base = Date.now() - count;
|
||||
for (let i = 0; i < count; i++) {
|
||||
const entryId = `${base + i}-seed${i}`;
|
||||
scored.push(base + i, entryId);
|
||||
fields.push(
|
||||
entryId,
|
||||
JSON.stringify({ event: event(`seed-${i}`), socketAttempts: 0 }),
|
||||
);
|
||||
}
|
||||
await redis.zadd(pendingKey(id), ...scored);
|
||||
await redis.hset(entriesKey(id), ...fields);
|
||||
await redis.zadd(INDEX_KEY, base, id);
|
||||
await redis.incrby(COUNTER_KEY, count);
|
||||
};
|
||||
|
||||
const pendingEvents = async (id = subId): Promise<ProjectedEvent[]> => {
|
||||
const held = await redis.hvals(entriesKey(id));
|
||||
return held.map(
|
||||
(raw: string) => (JSON.parse(raw) as { event: ProjectedEvent }).event,
|
||||
);
|
||||
};
|
||||
|
||||
beforeEach(async () => {
|
||||
seq++;
|
||||
subId = `app-x#sub-${seq}`;
|
||||
commands = [];
|
||||
redis = recordingRedis(new MockRedis.Cluster(['redis://localhost:7001']));
|
||||
await redis.del(INDEX_KEY, COUNTER_KEY);
|
||||
store = new PendingDeliveryStore(
|
||||
{} as IConfig,
|
||||
{ redis } as never,
|
||||
{} as never,
|
||||
);
|
||||
});
|
||||
|
||||
afterEach(() => {
|
||||
vi.useRealTimers();
|
||||
});
|
||||
|
||||
describe('what a subscription owns', () => {
|
||||
it('owns nothing until it is owed something', async () => {
|
||||
await expect(keysOf()).resolves.toEqual([]);
|
||||
await expect(store.depth(subId)).resolves.toBe(0);
|
||||
|
||||
await store.enqueue(subId, event('a'));
|
||||
|
||||
expect(await keysOf()).not.toEqual([]);
|
||||
await expect(store.depth(subId)).resolves.toBe(1);
|
||||
});
|
||||
|
||||
it('owns nothing again once the last delivery settles', async () => {
|
||||
const { entryId } = await store.enqueue(subId, event('a'));
|
||||
|
||||
await expect(store.settle(subId, entryId)).resolves.toBe(true);
|
||||
|
||||
await expect(keysOf()).resolves.toEqual([]);
|
||||
await expect(store.head(10)).resolves.toEqual([]);
|
||||
await expect(store.regionDepth()).resolves.toBe(0);
|
||||
});
|
||||
});
|
||||
|
||||
describe('handing a delivery out', () => {
|
||||
it('hands out the oldest first, and only one at a time', async () => {
|
||||
const first = await store.enqueue(subId, event('a'));
|
||||
await store.enqueue(subId, event('b'));
|
||||
|
||||
const claimed = await store.claim(subId);
|
||||
expect(claimed?.entryId).toBe(first.entryId);
|
||||
expect(claimed?.event.id).toBe('a');
|
||||
|
||||
// The second is owed, but nothing else may take it while the first is
|
||||
// out — `single` promises one consumer per event.
|
||||
await expect(store.claim(subId)).resolves.toBeNull();
|
||||
|
||||
await store.settle(subId, first.entryId);
|
||||
await expect(store.claim(subId)).resolves.toMatchObject({
|
||||
event: { id: 'b' },
|
||||
});
|
||||
});
|
||||
|
||||
it('counts the socket attempts a delivery has spent', async () => {
|
||||
const { entryId } = await store.enqueue(subId, event('a'));
|
||||
await store.claim(subId, { leaseMs: 1 });
|
||||
|
||||
await expect(store.recordSocketAttempt(subId, entryId)).resolves.toBe(
|
||||
1,
|
||||
);
|
||||
|
||||
vi.useFakeTimers({ toFake: ['Date'] });
|
||||
vi.setSystemTime(Date.now() + 1_000);
|
||||
await expect(store.claim(subId)).resolves.toMatchObject({
|
||||
entryId,
|
||||
socketAttempts: 1,
|
||||
});
|
||||
});
|
||||
|
||||
it('offers a delivery again once its lease lapses', async () => {
|
||||
const { entryId } = await store.enqueue(subId, event('a'));
|
||||
expect(await store.claim(subId, { leaseMs: 30_000 })).not.toBeNull();
|
||||
await expect(store.claim(subId)).resolves.toBeNull();
|
||||
|
||||
vi.useFakeTimers({ toFake: ['Date'] });
|
||||
vi.setSystemTime(Date.now() + 31_000);
|
||||
|
||||
await expect(store.claim(subId)).resolves.toMatchObject({ entryId });
|
||||
});
|
||||
|
||||
it('settles once, and treats a second ack as nothing to do', async () => {
|
||||
const { entryId } = await store.enqueue(subId, event('a'));
|
||||
await store.claim(subId);
|
||||
|
||||
await expect(store.settle(subId, entryId)).resolves.toBe(true);
|
||||
await expect(store.settle(subId, entryId)).resolves.toBe(false);
|
||||
});
|
||||
});
|
||||
|
||||
describe('when there is more than can be held', () => {
|
||||
it('drops the oldest of one backlog and leaves one marker', async () => {
|
||||
await seedBacklog(subId, EVENTS_PENDING_DELIVERIES_PER_SUBSCRIPTION);
|
||||
|
||||
const { shed } = await store.enqueue(subId, event('newest'));
|
||||
|
||||
expect(shed).toEqual([{ subId, dropped: 2, scope: 'subscription' }]);
|
||||
await expect(store.depth(subId)).resolves.toBe(
|
||||
EVENTS_PENDING_DELIVERIES_PER_SUBSCRIPTION,
|
||||
);
|
||||
|
||||
const held = await pendingEvents();
|
||||
const markers = held.filter((held) => held.op === 'gap');
|
||||
expect(markers).toHaveLength(1);
|
||||
expect(markers[0]).toMatchObject({
|
||||
op: 'gap',
|
||||
reason: 'backlog_overflow',
|
||||
subject: 'fs:/u/Documents',
|
||||
});
|
||||
// The oldest went, the newest stayed.
|
||||
expect(held.some((event) => event.id === 'seed-0')).toBe(false);
|
||||
expect(held.some((event) => event.id === 'newest')).toBe(true);
|
||||
});
|
||||
|
||||
it('sheds the region`s oldest backlog first, and says which', async () => {
|
||||
const older = `app-x#older-${seq}`;
|
||||
await seedBacklog(older, 4);
|
||||
// Two over the ceiling once this enqueue lands. Getting back under it
|
||||
// takes three, because the marker left behind holds a place too — and
|
||||
// the shed stops inside the oldest backlog, never reaching the newest.
|
||||
await redis.incrby(COUNTER_KEY, EVENTS_REGION_PENDING_CEILING - 3);
|
||||
|
||||
const { shed } = await store.enqueue(subId, event('a'));
|
||||
|
||||
expect(shed).toEqual([{ subId: older, dropped: 3, scope: 'region' }]);
|
||||
// What it lost, it was told about.
|
||||
const marked = await pendingEvents(older);
|
||||
expect(marked.filter((event) => event.op === 'gap')).toHaveLength(1);
|
||||
// What it did not lose is still owed; the newest end survives a shed.
|
||||
expect(marked.some((event) => event.id === 'seed-3')).toBe(true);
|
||||
expect(marked.some((event) => event.id === 'seed-2')).toBe(false);
|
||||
await expect(store.depth(subId)).resolves.toBe(1);
|
||||
// The region actually got back under, marker included.
|
||||
await expect(store.regionDepth()).resolves.toBe(
|
||||
EVENTS_REGION_PENDING_CEILING,
|
||||
);
|
||||
});
|
||||
});
|
||||
|
||||
describe('sharing the sweeper between backlogs', () => {
|
||||
beforeEach(() => {
|
||||
vi.useFakeTimers();
|
||||
});
|
||||
|
||||
it('hands the head to exactly one of two concurrent claimers', async () => {
|
||||
await store.enqueue(subId, event('a'));
|
||||
|
||||
const claims = await Promise.all([
|
||||
store.claim(subId),
|
||||
store.claim(subId),
|
||||
]);
|
||||
|
||||
expect(claims.filter(Boolean)).toHaveLength(1);
|
||||
});
|
||||
|
||||
it('moves a claimed subscription behind the ones still waiting', async () => {
|
||||
const other = `app-x#other-${seq}`;
|
||||
vi.setSystemTime(1_000_000);
|
||||
await store.enqueue(subId, event('first'));
|
||||
vi.setSystemTime(1_001_000);
|
||||
await store.enqueue(other, event('second'));
|
||||
expect((await store.head(2)).map((h) => h.subId)).toEqual([
|
||||
subId,
|
||||
other,
|
||||
]);
|
||||
|
||||
vi.setSystemTime(1_002_000);
|
||||
await store.claim(subId);
|
||||
|
||||
expect((await store.head(2)).map((h) => h.subId)).toEqual([
|
||||
other,
|
||||
subId,
|
||||
]);
|
||||
});
|
||||
|
||||
it('defers a subscription behind the others, holdings untouched', async () => {
|
||||
const other = `app-x#other-${seq}`;
|
||||
vi.setSystemTime(1_000_000);
|
||||
await store.enqueue(subId, event('first'));
|
||||
vi.setSystemTime(1_001_000);
|
||||
await store.enqueue(other, event('second'));
|
||||
|
||||
vi.setSystemTime(1_002_000);
|
||||
await store.defer(subId);
|
||||
|
||||
expect((await store.head(2)).map((h) => h.subId)).toEqual([
|
||||
other,
|
||||
subId,
|
||||
]);
|
||||
await expect(store.depth(subId)).resolves.toBe(1);
|
||||
});
|
||||
|
||||
it('puts a lifetime on a backlog`s keys', async () => {
|
||||
await store.enqueue(subId, event('a'));
|
||||
|
||||
await expect(redis.ttl(pendingKey())).resolves.toBeGreaterThan(0);
|
||||
await expect(redis.ttl(entriesKey())).resolves.toBeGreaterThan(0);
|
||||
});
|
||||
});
|
||||
|
||||
describe('finding the work', () => {
|
||||
it('reads the oldest end of the index rather than the keyspace', async () => {
|
||||
const older = `app-x#older-${seq}`;
|
||||
await seedBacklog(older, 2);
|
||||
await store.enqueue(subId, event('a'));
|
||||
|
||||
commands = [];
|
||||
const head = await store.head(10);
|
||||
|
||||
expect(head.map((entry) => entry.subId)).toEqual([older, subId]);
|
||||
expect(head[0].oldestAt).toBeLessThan(head[1].oldestAt);
|
||||
expect(
|
||||
commands.filter((command) =>
|
||||
['scan', 'keys', 'hscan', 'sscan', 'zscan'].includes(command),
|
||||
),
|
||||
).toEqual([]);
|
||||
});
|
||||
|
||||
it('never scans the keyspace, whatever it is asked to do', async () => {
|
||||
const { entryId } = await store.enqueue(subId, event('a'));
|
||||
await store.claim(subId);
|
||||
await store.recordSocketAttempt(subId, entryId);
|
||||
await store.settle(subId, entryId);
|
||||
await store.enqueue(subId, event('b'));
|
||||
await store.purge(subId);
|
||||
await store.regionDepth();
|
||||
|
||||
expect(
|
||||
commands.filter((command) =>
|
||||
['scan', 'keys', 'hscan', 'sscan', 'zscan'].includes(command),
|
||||
),
|
||||
).toEqual([]);
|
||||
});
|
||||
});
|
||||
|
||||
describe('surviving a crash between the write and the reindex', () => {
|
||||
it('keeps a first entry discoverable even if the follow-up reindex never runs', async () => {
|
||||
// The counter write is the first command after the entry's own
|
||||
// transaction lands — exactly where a crash would fall between the
|
||||
// entry existing and the follow-up reindex.
|
||||
vi.spyOn(redis, 'incrby').mockRejectedValueOnce(
|
||||
new Error('connection lost'),
|
||||
);
|
||||
|
||||
await expect(store.enqueue(subId, event('a'))).rejects.toThrow();
|
||||
vi.restoreAllMocks();
|
||||
|
||||
// The crash was after the index write, not before it: the sweeper
|
||||
// still finds this subscription, and the entry is still claimable.
|
||||
await expect(store.head(10)).resolves.toEqual([
|
||||
{ subId, oldestAt: expect.any(Number) },
|
||||
]);
|
||||
await expect(store.claim(subId)).resolves.toMatchObject({
|
||||
event: { id: 'a' },
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
describe('keeping the region counter honest', () => {
|
||||
it('corrects an inflated counter back to what is actually held', async () => {
|
||||
await store.enqueue(subId, event('a'));
|
||||
// Drift: some decrement that should have landed did not.
|
||||
await redis.incrby(COUNTER_KEY, 500);
|
||||
|
||||
await expect(store.reconcileRegionDepth()).resolves.toBe(1);
|
||||
await expect(store.regionDepth()).resolves.toBe(1);
|
||||
});
|
||||
|
||||
it('corrects an undercounted or negative counter the same way', async () => {
|
||||
await store.enqueue(subId, event('a'));
|
||||
await redis.set(COUNTER_KEY, -50);
|
||||
|
||||
await expect(store.reconcileRegionDepth()).resolves.toBe(1);
|
||||
await expect(store.regionDepth()).resolves.toBe(1);
|
||||
});
|
||||
|
||||
it('zeroes the counter when nothing is actually pending', async () => {
|
||||
await redis.incrby(COUNTER_KEY, 12);
|
||||
|
||||
await expect(store.reconcileRegionDepth()).resolves.toBe(0);
|
||||
await expect(store.regionDepth()).resolves.toBe(0);
|
||||
});
|
||||
|
||||
it('never scans the keyspace to do it', async () => {
|
||||
await store.enqueue(subId, event('a'));
|
||||
commands = [];
|
||||
|
||||
await store.reconcileRegionDepth();
|
||||
|
||||
expect(
|
||||
commands.filter((command) =>
|
||||
['scan', 'keys', 'hscan', 'sscan', 'zscan'].includes(command),
|
||||
),
|
||||
).toEqual([]);
|
||||
});
|
||||
});
|
||||
|
||||
describe('giving up a backlog', () => {
|
||||
it('drops everything a subscription held and gives the room back', async () => {
|
||||
await store.enqueue(subId, event('a'));
|
||||
await store.enqueue(subId, event('b'));
|
||||
await expect(store.regionDepth()).resolves.toBe(2);
|
||||
|
||||
await expect(store.purge(subId)).resolves.toBe(2);
|
||||
|
||||
await expect(keysOf()).resolves.toEqual([]);
|
||||
await expect(store.regionDepth()).resolves.toBe(0);
|
||||
await expect(store.head(10)).resolves.toEqual([]);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,590 @@
|
||||
/*
|
||||
* Copyright (C) 2024-present Puter Technologies Inc.
|
||||
*
|
||||
* This file is part of Puter.
|
||||
*
|
||||
* Puter is free software: you can redistribute it and/or modify
|
||||
* it under the terms of the GNU Affero General Public License as published
|
||||
* by the Free Software Foundation, either version 3 of the License, or
|
||||
* (at your option) any later version.
|
||||
*
|
||||
* This program is distributed in the hope that it will be useful,
|
||||
* but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
* GNU Affero General Public License for more details.
|
||||
*
|
||||
* You should have received a copy of the GNU Affero General Public License
|
||||
* along with this program. If not, see <https://www.gnu.org/licenses/>.
|
||||
*/
|
||||
|
||||
import { randomUUID } from 'node:crypto';
|
||||
import {
|
||||
EVENTS_PENDING_DELIVERIES_PER_SUBSCRIPTION,
|
||||
EVENTS_REGION_PENDING_CEILING,
|
||||
} from '../../controllers/events/limits.js';
|
||||
import type {
|
||||
DeliverableEvent,
|
||||
GapMarker,
|
||||
} from '../../services/events/registry.js';
|
||||
import { PuterStore } from '../types.js';
|
||||
|
||||
/**
|
||||
* Deliveries that are waiting for a consumer, in the region that emitted them.
|
||||
*
|
||||
* A `broadcast` delivery is gone the moment it is sent. A `single` one is owed
|
||||
* to exactly one consumer, so it has to survive until that consumer says it
|
||||
* took it — which is what everything here holds. Nothing is replicated: the
|
||||
* lease, the retry count and the queue belong to the region that emitted the
|
||||
* event, and the published promise is at-least-once **while that region is
|
||||
* available**.
|
||||
*
|
||||
* Three keys per subscription, all created on the first pending delivery and
|
||||
* all deleted the moment the last one settles — a subscription that is keeping
|
||||
* up owns nothing:
|
||||
*
|
||||
* ev:q:{<subId>} HASH entryId -> the delivery and its attempt count
|
||||
* ev:qp:{<subId>} ZSET entryId -> enqueued at; membership means unsettled
|
||||
* ev:ql:{<subId>} ZSET entryId -> lease expiry; membership means in flight
|
||||
*
|
||||
* And two the region shares:
|
||||
*
|
||||
* ev:qx ZSET subId -> oldest pending delivery, for the sweeper
|
||||
* ev:qc STR how many deliveries the region is holding
|
||||
*
|
||||
* `ev:qx` is why nothing here ever scans the keyspace: the sweeper reads its
|
||||
* head to find the subscriptions that are behind, and a scan is exactly what
|
||||
* goes pathological when the system already is. The counter is what the region
|
||||
* ceiling is read from, and it moves only where the pending set does.
|
||||
*
|
||||
* The pending set, the lease and the attempt count are explicit rather than a
|
||||
* queue primitive's implicit ones: reclaiming an expired lease is then a score
|
||||
* comparison, and dropping a whole subscription is one delete.
|
||||
*
|
||||
* One delivery is in flight per subscription at a time. `single` promises one
|
||||
* consumer per event, and handing out the next while the last is unsettled
|
||||
* would make ordering — and the retry count that decides socket-versus-worker —
|
||||
* meaningless.
|
||||
*/
|
||||
|
||||
// -- Keys -------------------------------------------------------------
|
||||
|
||||
const entriesKey = (subId: string): string => `ev:q:{${subId}}`;
|
||||
const pendingKey = (subId: string): string => `ev:qp:{${subId}}`;
|
||||
const leaseKey = (subId: string): string => `ev:ql:{${subId}}`;
|
||||
|
||||
const INDEX_KEY = 'ev:qx';
|
||||
const COUNTER_KEY = 'ev:qc';
|
||||
|
||||
// -- Lease ------------------------------------------------------------
|
||||
|
||||
/**
|
||||
* How long a consumer has to settle a delivery before someone else may take it.
|
||||
* Long enough for a handler doing real work, short enough that a tab that
|
||||
* closed mid-delivery does not hold the queue.
|
||||
*/
|
||||
export const PENDING_LEASE_TTL_MS = 30_000;
|
||||
|
||||
/** Subscriptions one region-ceiling shed may take deliveries from. */
|
||||
const REGION_SHED_SUBSCRIPTIONS = 32;
|
||||
|
||||
/** Deliveries one region-ceiling shed may drop, so a burst cannot stall here. */
|
||||
const REGION_SHED_MAX_ENTRIES = 1_000;
|
||||
|
||||
/** Arguments one variadic command carries, so a large shed stays in batches. */
|
||||
const COMMAND_BATCH = 500;
|
||||
|
||||
/** Queues the reconciler measures at once, one command each. */
|
||||
const RECONCILE_CONCURRENCY = 50;
|
||||
|
||||
/**
|
||||
* Backstop for keys nothing indexes any more: a claim refreshes it, so a
|
||||
* backlog still being retried never lapses, while one left behind by a purge
|
||||
* that raced an enqueue does not sit in Redis forever.
|
||||
*/
|
||||
export const PENDING_BACKLOG_TTL_SECONDS = 7 * 24 * 60 * 60;
|
||||
|
||||
// -- Scripts ----------------------------------------------------------
|
||||
// The three keys a subscription owns share its `{subId}` hash tag, so a script
|
||||
// over them runs on one node under cluster mode. The index and the counter hash
|
||||
// elsewhere and are moved by plain commands around the scripts.
|
||||
|
||||
/**
|
||||
* Lease the oldest delivery, if none is in flight. KEYS: entries, pending,
|
||||
* lease. ARGV: now, leaseUntil, ttlSeconds. Returns a status-first tuple.
|
||||
*/
|
||||
const CLAIM_SCRIPT = `
|
||||
local inflight = redis.call('ZRANGEBYSCORE', KEYS[3], ARGV[1], '+inf', 'LIMIT', 0, 1)
|
||||
if #inflight > 0 then return { 'inflight' } end
|
||||
local head = redis.call('ZRANGE', KEYS[2], 0, 0)
|
||||
if #head == 0 then return { 'empty' } end
|
||||
local raw = redis.call('HGET', KEYS[1], head[1])
|
||||
if not raw then
|
||||
redis.call('ZREM', KEYS[2], head[1])
|
||||
redis.call('ZREM', KEYS[3], head[1])
|
||||
return { 'missing', head[1] }
|
||||
end
|
||||
redis.call('ZADD', KEYS[3], ARGV[2], head[1])
|
||||
redis.call('EXPIRE', KEYS[1], ARGV[3])
|
||||
redis.call('EXPIRE', KEYS[2], ARGV[3])
|
||||
redis.call('EXPIRE', KEYS[3], ARGV[3])
|
||||
return { 'claimed', head[1], raw }
|
||||
`;
|
||||
|
||||
/**
|
||||
* The oldest pending score, or '-1' after deleting a drained subscription's
|
||||
* keys. Checking and deleting in one step is what keeps a concurrent append
|
||||
* from being wiped between the two. KEYS: entries, pending, lease.
|
||||
*/
|
||||
const REINDEX_SCRIPT = `
|
||||
local head = redis.call('ZRANGE', KEYS[2], 0, 0, 'WITHSCORES')
|
||||
if #head == 0 then
|
||||
redis.call('DEL', KEYS[1], KEYS[2], KEYS[3])
|
||||
return '-1'
|
||||
end
|
||||
return head[2]
|
||||
`;
|
||||
|
||||
interface PendingScripts {
|
||||
pendingClaim(
|
||||
entries: string,
|
||||
pending: string,
|
||||
lease: string,
|
||||
now: string,
|
||||
leaseUntil: string,
|
||||
ttlSeconds: string,
|
||||
): Promise<[string, string?, string?]>;
|
||||
pendingReindex(
|
||||
entries: string,
|
||||
pending: string,
|
||||
lease: string,
|
||||
): Promise<string>;
|
||||
}
|
||||
|
||||
// -- Shapes -----------------------------------------------------------
|
||||
|
||||
/** One delivery, handed out under a lease. */
|
||||
export interface ClaimedDelivery {
|
||||
entryId: string;
|
||||
event: DeliverableEvent;
|
||||
/** Socket attempts already spent, which is what decides the next one. */
|
||||
socketAttempts: number;
|
||||
}
|
||||
|
||||
/** What one shed took, for the marker and the alarm that follow it. */
|
||||
export interface PendingShed {
|
||||
subId: string;
|
||||
dropped: number;
|
||||
scope: 'subscription' | 'region';
|
||||
}
|
||||
|
||||
/** A subscription with undelivered deliveries, oldest first. */
|
||||
export interface PendingHead {
|
||||
subId: string;
|
||||
oldestAt: number;
|
||||
}
|
||||
|
||||
interface StoredEntry {
|
||||
event: DeliverableEvent;
|
||||
socketAttempts: number;
|
||||
}
|
||||
|
||||
const parseEntry = (raw: string | null): StoredEntry | null => {
|
||||
if (!raw) return null;
|
||||
try {
|
||||
const parsed = JSON.parse(raw) as StoredEntry;
|
||||
return parsed?.event ? parsed : null;
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
};
|
||||
|
||||
const gapMarker = (subject: string): GapMarker => ({
|
||||
id: randomUUID(),
|
||||
subject,
|
||||
op: 'gap',
|
||||
reason: 'backlog_overflow',
|
||||
ts: Date.now(),
|
||||
});
|
||||
|
||||
/** `ZPOPMIN`/`ZRANGE … WITHSCORES` answer flat, and scores come back typed. */
|
||||
const membersOf = (flat: readonly unknown[]): string[] => {
|
||||
const members: string[] = [];
|
||||
for (let i = 0; i < flat.length; i += 2) members.push(String(flat[i]));
|
||||
return members;
|
||||
};
|
||||
|
||||
const batched = <T>(items: readonly T[], size = COMMAND_BATCH): T[][] => {
|
||||
const batches: T[][] = [];
|
||||
for (let i = 0; i < items.length; i += size)
|
||||
batches.push(items.slice(i, i + size));
|
||||
return batches;
|
||||
};
|
||||
|
||||
export class PendingDeliveryStore extends PuterStore {
|
||||
/** Tells this process's ids from a peer's, so two cannot mint the same. */
|
||||
readonly #minter = randomUUID().slice(0, 8);
|
||||
#minted = 0;
|
||||
#definedScripts = false;
|
||||
|
||||
#scripts(): PendingScripts {
|
||||
if (!this.#definedScripts) {
|
||||
this.#definedScripts = true;
|
||||
this.clients.redis.defineCommand('pendingClaim', {
|
||||
numberOfKeys: 3,
|
||||
lua: CLAIM_SCRIPT,
|
||||
});
|
||||
this.clients.redis.defineCommand('pendingReindex', {
|
||||
numberOfKeys: 3,
|
||||
lua: REINDEX_SCRIPT,
|
||||
});
|
||||
}
|
||||
return this.clients.redis as unknown as PendingScripts;
|
||||
}
|
||||
|
||||
// -- Writes ------------------------------------------------------
|
||||
|
||||
/**
|
||||
* Hold one delivery until a consumer takes it. Returns what had to be shed
|
||||
* to make room — the subscription's own cap first, then the region's — so
|
||||
* the caller can say so out loud.
|
||||
*/
|
||||
async enqueue(
|
||||
subId: string,
|
||||
event: DeliverableEvent,
|
||||
): Promise<{ entryId: string; shed: PendingShed[] }> {
|
||||
const entryId = await this.#append(subId, event);
|
||||
|
||||
const shed: PendingShed[] = [];
|
||||
const overflowed = await this.#capSubscription(subId);
|
||||
if (overflowed) shed.push(overflowed);
|
||||
shed.push(...(await this.#capRegion()));
|
||||
|
||||
return { entryId, shed };
|
||||
}
|
||||
|
||||
/**
|
||||
* Take the oldest delivery this subscription is owed, under a lease. Null
|
||||
* when one is already in flight or nothing is waiting; an expired lease
|
||||
* makes its delivery claimable again, which is the whole retry mechanism.
|
||||
*/
|
||||
async claim(
|
||||
subId: string,
|
||||
options: { leaseMs?: number } = {},
|
||||
): Promise<ClaimedDelivery | null> {
|
||||
const now = Date.now();
|
||||
// One script, so two claimers racing for the same head cannot both
|
||||
// walk away holding it.
|
||||
const [status, entryId, raw] = await this.#scripts().pendingClaim(
|
||||
entriesKey(subId),
|
||||
pendingKey(subId),
|
||||
leaseKey(subId),
|
||||
String(now),
|
||||
String(now + (options.leaseMs ?? PENDING_LEASE_TTL_MS)),
|
||||
String(PENDING_BACKLOG_TTL_SECONDS),
|
||||
);
|
||||
if (status === 'inflight') return null;
|
||||
if (status === 'empty') {
|
||||
// Nothing left, which is also how a drained subscription's keys go.
|
||||
await this.#reindex(subId);
|
||||
return null;
|
||||
}
|
||||
if (status === 'missing') {
|
||||
// A queue position with no entry behind it: half a write. The
|
||||
// script already dropped the position; this drops its share.
|
||||
await this.clients.redis.decrby(COUNTER_KEY, 1);
|
||||
await this.#reindex(subId);
|
||||
return null;
|
||||
}
|
||||
|
||||
const entry = parseEntry(raw ?? null);
|
||||
if (!entryId || !entry) {
|
||||
await this.clients.redis.zrem(pendingKey(subId), String(entryId));
|
||||
await this.#forget(subId, [String(entryId)]);
|
||||
await this.#reindex(subId);
|
||||
return null;
|
||||
}
|
||||
|
||||
// To the back of the sweeper's line: a delivery nobody ever settles
|
||||
// must not hold the head against every other backlog in the region.
|
||||
await this.clients.redis.zadd(INDEX_KEY, 'XX', now, subId);
|
||||
return {
|
||||
entryId,
|
||||
event: entry.event,
|
||||
socketAttempts: entry.socketAttempts,
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Send a subscription to the back of the sweeper's line, holdings
|
||||
* untouched.
|
||||
*/
|
||||
async defer(subId: string): Promise<void> {
|
||||
await this.clients.redis.zadd(INDEX_KEY, 'XX', Date.now(), subId);
|
||||
}
|
||||
|
||||
/** Count one socket attempt against a claimed delivery. */
|
||||
async recordSocketAttempt(subId: string, entryId: string): Promise<number> {
|
||||
const entry = parseEntry(
|
||||
await this.clients.redis.hget(entriesKey(subId), entryId),
|
||||
);
|
||||
if (!entry) return 0;
|
||||
|
||||
const socketAttempts = entry.socketAttempts + 1;
|
||||
await this.clients.redis.hset(
|
||||
entriesKey(subId),
|
||||
entryId,
|
||||
JSON.stringify({ ...entry, socketAttempts }),
|
||||
);
|
||||
return socketAttempts;
|
||||
}
|
||||
|
||||
/**
|
||||
* Settle a delivery a consumer took. False for an id this subscription is
|
||||
* not holding — a second ack for one already settled, which at-least-once
|
||||
* makes routine and which nothing should treat as an error.
|
||||
*/
|
||||
async settle(subId: string, entryId: string): Promise<boolean> {
|
||||
const removed = await this.clients.redis.zrem(
|
||||
pendingKey(subId),
|
||||
entryId,
|
||||
);
|
||||
if (Number(removed) !== 1) return false;
|
||||
|
||||
await this.#forget(subId, [entryId]);
|
||||
await this.#reindex(subId);
|
||||
return true;
|
||||
}
|
||||
|
||||
/** Drop everything a subscription is holding, for one that is going away. */
|
||||
async purge(subId: string): Promise<number> {
|
||||
const held = await this.depth(subId);
|
||||
await this.clients.redis.del(
|
||||
entriesKey(subId),
|
||||
pendingKey(subId),
|
||||
leaseKey(subId),
|
||||
);
|
||||
await this.clients.redis.zrem(INDEX_KEY, subId);
|
||||
if (held > 0) await this.clients.redis.decrby(COUNTER_KEY, held);
|
||||
return held;
|
||||
}
|
||||
|
||||
// -- Reads -------------------------------------------------------
|
||||
|
||||
/**
|
||||
* The subscriptions that have waited longest, oldest first. The sweeper's
|
||||
* whole input, and the backlog-age metric's.
|
||||
*/
|
||||
async head(limit: number): Promise<PendingHead[]> {
|
||||
if (limit <= 0) return [];
|
||||
const flat = await this.clients.redis.zrange(
|
||||
INDEX_KEY,
|
||||
0,
|
||||
limit - 1,
|
||||
'WITHSCORES',
|
||||
);
|
||||
const heads: PendingHead[] = [];
|
||||
for (let i = 0; i < flat.length; i += 2)
|
||||
heads.push({
|
||||
subId: String(flat[i]),
|
||||
oldestAt: Number(flat[i + 1]) || 0,
|
||||
});
|
||||
return heads;
|
||||
}
|
||||
|
||||
async depth(subId: string): Promise<number> {
|
||||
return Number(await this.clients.redis.zcard(pendingKey(subId))) || 0;
|
||||
}
|
||||
|
||||
/** What the region is holding in total, and what its ceiling is read from. */
|
||||
async regionDepth(): Promise<number> {
|
||||
const raw = await this.clients.redis.get(COUNTER_KEY);
|
||||
const held = raw === null ? 0 : Number.parseInt(raw, 10);
|
||||
return Number.isFinite(held) && held > 0 ? held : 0;
|
||||
}
|
||||
|
||||
/**
|
||||
* Recompute the region counter from the pending sets it is supposed to
|
||||
* total. The counter moves in a separate write from the removal it follows
|
||||
* (a settle's decrement, a shed's decrement-then-append), so a crash
|
||||
* between the two drifts it in either direction — silently undercounting
|
||||
* lets the ceiling never trip, silently overcounting trips it forever.
|
||||
* Cheap in the steady state the index is sized for: one read per
|
||||
* subscription that actually has a backlog, never the keyspace.
|
||||
*/
|
||||
async reconcileRegionDepth(): Promise<number> {
|
||||
const subIds = await this.clients.redis.zrange(INDEX_KEY, 0, -1);
|
||||
|
||||
let total = 0;
|
||||
// One command per subscription: the queues hash to different slots,
|
||||
// so under cluster mode they cannot share a pipeline.
|
||||
for (const batch of batched(subIds, RECONCILE_CONCURRENCY)) {
|
||||
const counts = await Promise.all(
|
||||
batch.map((subId) => this.depth(subId)),
|
||||
);
|
||||
for (const count of counts) total += count;
|
||||
}
|
||||
|
||||
await this.clients.redis.set(COUNTER_KEY, total);
|
||||
return total;
|
||||
}
|
||||
|
||||
// -- Internals ---------------------------------------------------
|
||||
|
||||
/**
|
||||
* The id and the order of one delivery. Deliveries landing in the same
|
||||
* millisecond share a score, and a sorted set breaks that tie on the member
|
||||
* — so the counter is padded and comes first, and a queue keeps its order
|
||||
* under a burst instead of shuffling it.
|
||||
*/
|
||||
#mintEntryId(at: number): string {
|
||||
const minted = String(++this.#minted).padStart(12, '0');
|
||||
return `${at}-${minted}-${this.#minter}`;
|
||||
}
|
||||
|
||||
/**
|
||||
* Add one delivery without asking whether there is room for it.
|
||||
*
|
||||
* The index write goes in the same pipeline as the entry it describes —
|
||||
* `NX` so it never disturbs an existing (older) score — rather than waiting
|
||||
* for the follow-up `#reindex` below. A subscription's first pending entry
|
||||
* would otherwise be fully written and claimable, yet invisible to the
|
||||
* sweeper forever, if this process died in the gap between the two: `ev:qx`
|
||||
* is the only thing the sweeper ever reads, so an entry it does not know
|
||||
* about is never retried.
|
||||
*/
|
||||
async #append(subId: string, event: DeliverableEvent): Promise<string> {
|
||||
const now = Date.now();
|
||||
const entryId = this.#mintEntryId(now);
|
||||
|
||||
// Index first, so a crash after the entry lands still leaves the
|
||||
// sweeper a way to find it; `NX` keeps an older score in place.
|
||||
await this.clients.redis.zadd(INDEX_KEY, 'NX', now, subId);
|
||||
// The entry and its queue position land together, so a concurrent
|
||||
// drain can never see one without the other. Same slot, so this is a
|
||||
// real transaction under cluster mode.
|
||||
const write = this.clients.redis.multi();
|
||||
write.hset(
|
||||
entriesKey(subId),
|
||||
entryId,
|
||||
JSON.stringify({ event, socketAttempts: 0 } satisfies StoredEntry),
|
||||
);
|
||||
write.zadd(pendingKey(subId), now, entryId);
|
||||
write.expire(entriesKey(subId), PENDING_BACKLOG_TTL_SECONDS);
|
||||
write.expire(pendingKey(subId), PENDING_BACKLOG_TTL_SECONDS);
|
||||
await write.exec();
|
||||
await this.clients.redis.incrby(COUNTER_KEY, 1);
|
||||
|
||||
// A drain that emptied this subscription in between took it out of
|
||||
// the index again; this puts it back, with the right score.
|
||||
await this.#reindex(subId);
|
||||
return entryId;
|
||||
}
|
||||
|
||||
async #capSubscription(subId: string): Promise<PendingShed | null> {
|
||||
const held = await this.depth(subId);
|
||||
const over = held - EVENTS_PENDING_DELIVERIES_PER_SUBSCRIPTION;
|
||||
if (over <= 0) return null;
|
||||
// One more than the overflow, because the marker that replaces them
|
||||
// takes a place of its own.
|
||||
return this.#shedOldest(subId, over + 1, 'subscription');
|
||||
}
|
||||
|
||||
/**
|
||||
* Shed the region's oldest deliveries until it is back under the ceiling.
|
||||
* Oldest-first across subscriptions, because the index is already ordered
|
||||
* that way and a backlog nobody is draining is the one to lose.
|
||||
*/
|
||||
async #capRegion(): Promise<PendingShed[]> {
|
||||
const held = await this.regionDepth();
|
||||
let over = Math.min(
|
||||
held - EVENTS_REGION_PENDING_CEILING,
|
||||
REGION_SHED_MAX_ENTRIES,
|
||||
);
|
||||
if (over <= 0) return [];
|
||||
|
||||
const shed: PendingShed[] = [];
|
||||
for (const { subId } of await this.head(REGION_SHED_SUBSCRIPTIONS)) {
|
||||
if (over <= 0) break;
|
||||
// One more than the overflow, as the marker left behind takes a
|
||||
// place of its own; counting it is what lets the region actually
|
||||
// get back under the ceiling rather than hover one over it.
|
||||
const dropped = await this.#shedOldest(subId, over + 1, 'region');
|
||||
if (!dropped) continue;
|
||||
over -= dropped.dropped - 1;
|
||||
shed.push(dropped);
|
||||
}
|
||||
return shed;
|
||||
}
|
||||
|
||||
/**
|
||||
* Drop a subscription's oldest deliveries and leave one gap marker in their
|
||||
* place — the marker is a delivery like any other, so it queues where they
|
||||
* were and reaches the subscriber the same way.
|
||||
*/
|
||||
async #shedOldest(
|
||||
subId: string,
|
||||
count: number,
|
||||
scope: PendingShed['scope'],
|
||||
): Promise<PendingShed | null> {
|
||||
const dropped = membersOf(
|
||||
await this.clients.redis.zpopmin(pendingKey(subId), count),
|
||||
);
|
||||
if (dropped.length === 0) return null;
|
||||
|
||||
const subject = await this.#subjectOf(
|
||||
subId,
|
||||
dropped[dropped.length - 1],
|
||||
);
|
||||
await this.#forget(subId, dropped);
|
||||
await this.#append(subId, gapMarker(subject));
|
||||
|
||||
return { subId, dropped: dropped.length, scope };
|
||||
}
|
||||
|
||||
/** The subject a shed delivery carried, so its marker can name it. */
|
||||
async #subjectOf(subId: string, entryId: string): Promise<string> {
|
||||
const entry = parseEntry(
|
||||
await this.clients.redis.hget(entriesKey(subId), entryId),
|
||||
);
|
||||
return entry?.event.subject ?? '';
|
||||
}
|
||||
|
||||
/** Forget entries already out of the pending set, and the space they held. */
|
||||
async #forget(subId: string, entryIds: readonly string[]): Promise<void> {
|
||||
if (entryIds.length === 0) return;
|
||||
for (const batch of batched(entryIds)) {
|
||||
const drop = this.clients.redis.pipeline();
|
||||
drop.hdel(entriesKey(subId), ...batch);
|
||||
drop.zrem(leaseKey(subId), ...batch);
|
||||
await drop.exec();
|
||||
}
|
||||
await this.clients.redis.decrby(COUNTER_KEY, entryIds.length);
|
||||
}
|
||||
|
||||
/**
|
||||
* Point the sweeper's index at this subscription's oldest delivery, or take
|
||||
* the subscription out of the region entirely once it has none — which is
|
||||
* where every key it owned goes.
|
||||
*/
|
||||
async #reindex(subId: string): Promise<void> {
|
||||
const oldest = await this.#scripts().pendingReindex(
|
||||
entriesKey(subId),
|
||||
pendingKey(subId),
|
||||
leaseKey(subId),
|
||||
);
|
||||
if (oldest !== '-1') {
|
||||
await this.clients.redis.zadd(
|
||||
INDEX_KEY,
|
||||
Number(oldest) || 0,
|
||||
subId,
|
||||
);
|
||||
return;
|
||||
}
|
||||
await this.clients.redis.zrem(INDEX_KEY, subId);
|
||||
// The index hashes to another slot, so no script can cover both it and
|
||||
// the queue: an append that landed since the script ran has entries
|
||||
// the ZREM just hid from the sweeper. Put it back if so.
|
||||
if ((await this.depth(subId)) > 0)
|
||||
await this.clients.redis.zadd(INDEX_KEY, 'NX', Date.now(), subId);
|
||||
}
|
||||
}
|
||||
@@ -39,6 +39,30 @@ export const isSubscriptionTarget = (
|
||||
): value is SubscriptionTarget =>
|
||||
SUBSCRIPTION_TARGETS.includes(value as SubscriptionTarget);
|
||||
|
||||
/** Transports a session row may take: it has one connection and no handler. */
|
||||
export const SESSION_TARGETS: SubscriptionTarget[] = ['socket'];
|
||||
|
||||
/** Transports a durable row takes unless the caller says otherwise. */
|
||||
export const DEFAULT_DURABLE_TARGETS: SubscriptionTarget[] = [
|
||||
'socket',
|
||||
'worker',
|
||||
];
|
||||
|
||||
/**
|
||||
* Whether a delivery class may carry these transports. `single` and `push` are
|
||||
* incompatible by construction — a lease can only be settled by a consumer that
|
||||
* reports back, and a device notification never does — and `single` needs a
|
||||
* `worker` to fall back to once the connected clients have had their turns, or
|
||||
* an unacknowledged delivery sits at the head of its queue for good. Both are
|
||||
* invariants every writer is held to, not defaults.
|
||||
*/
|
||||
export const targetsAllowedForDelivery = (
|
||||
delivery: DeliveryClass,
|
||||
targets: readonly SubscriptionTarget[],
|
||||
): boolean =>
|
||||
delivery !== 'single' ||
|
||||
(!targets.includes('push') && targets.includes('worker'));
|
||||
|
||||
/** What dispatch needs from a subscription, whichever store it came from. */
|
||||
export interface DispatchSubscription {
|
||||
subId: string;
|
||||
@@ -59,13 +83,16 @@ export interface DispatchSubscription {
|
||||
appUid: string | null;
|
||||
/** ACL mode the subscribe check passed under; re-checked per delivery. */
|
||||
permission: AclMode;
|
||||
/** Transports this row's deliveries may take. */
|
||||
targets?: SubscriptionTarget[];
|
||||
/** Session rows only: the connection a delivery is addressed at. */
|
||||
socketId?: string;
|
||||
/** Durable rows only: set on every row that outlives its connection. */
|
||||
durable?: true;
|
||||
delivery?: DeliveryClass;
|
||||
targets?: SubscriptionTarget[];
|
||||
handlerName?: string | null;
|
||||
/** Durable rows only: handed to the handler, and read nowhere else. */
|
||||
context?: string | null;
|
||||
}
|
||||
|
||||
export interface SessionSubscription extends DispatchSubscription {
|
||||
|
||||
@@ -23,6 +23,7 @@ import { FSEntryStore } from './fs/FSEntryStore.js';
|
||||
import { GroupStore } from './group/GroupStore.js';
|
||||
import { DurableSubscriptionStore } from './events/DurableSubscriptionStore.js';
|
||||
import { EventSubscriptionStore } from './events/EventSubscriptionStore.js';
|
||||
import { PendingDeliveryStore } from './events/PendingDeliveryStore.js';
|
||||
import { CreditHoldStore } from './metering/CreditHoldStore.js';
|
||||
import { MeteringBufferStore } from './metering/MeteringBufferStore.js';
|
||||
import { NotificationStore } from './notification/NotificationStore.js';
|
||||
@@ -64,6 +65,7 @@ declare module './types.js' {
|
||||
userBlock: UserBlockStore;
|
||||
eventSubscription: EventSubscriptionStore;
|
||||
durableSubscription: DurableSubscriptionStore;
|
||||
pendingDelivery: PendingDeliveryStore;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -95,6 +97,7 @@ export const puterStores = {
|
||||
userBlock: UserBlockStore,
|
||||
// Redis only, no peer stores.
|
||||
eventSubscription: EventSubscriptionStore,
|
||||
pendingDelivery: PendingDeliveryStore,
|
||||
// Writes through the Redis keyspace above, so it comes after it.
|
||||
durableSubscription: DurableSubscriptionStore,
|
||||
} satisfies IPuterStoreRegistry;
|
||||
|
||||
@@ -168,6 +168,8 @@ One write can reach many subscriptions, so events are bounded on both halves: ho
|
||||
| Matched subscriptions per event | 50 |
|
||||
| Filter evaluations per event | 200 |
|
||||
| Deliveries per minute, per subscription | 600 |
|
||||
| Acknowledgements per minute | 600 |
|
||||
| Undelivered deliveries per subscription | 10,000 |
|
||||
|
||||
Subscriptions come in two kinds. A **session** subscription lives with the connection that made it: it is dropped when the connection closes, and a reconnecting client subscribes again. A **durable** subscription outlives every connection — it is created over the API, listed and revoked from the account, and keeps delivering until you remove it or it expires.
|
||||
|
||||
@@ -181,6 +183,8 @@ Match patterns are compiled once when you subscribe and are capped at **256 char
|
||||
|
||||
The three per-event ceilings do not fail your call — they truncate the delivery and send a `gap` marker in its place, an event with `op: 'gap'` and no `uid` or `path`. A gap means something happened that you were not told the details of, so a client that must not miss changes should re-read the anchor when it sees one rather than treat the silence as "nothing changed".
|
||||
|
||||
A `single` subscription is delivered to exactly one consumer, which has **30 seconds** to acknowledge each delivery before it is offered again — twice to a connected client, then to the subscription's handler. Until it is acknowledged it is held for you, so a consumer that is away is a backlog that grows: **10,000** undelivered deliveries per subscription, after which the oldest are dropped and one `gap` marker with `reason: 'backlog_overflow'` takes their place. Each region also holds at most **1,000,000** undelivered deliveries across every subscription it serves, and sheds the oldest first — with the same marker — before it reaches that. A redelivery after a missed acknowledgement is normal and expected: deliveries are at-least-once, `event.id` is stable across them, and a handler that runs twice on the same id should do nothing the second time.
|
||||
|
||||
### Peer connections
|
||||
|
||||
| Limit | Paid | Free | Anonymous |
|
||||
|
||||
@@ -43,8 +43,9 @@
|
||||
* @property {string} subject The subject that was being delivered.
|
||||
* @property {'gap'} op Always `'gap'`.
|
||||
* @property {string} reason Why the delivery was dropped —
|
||||
* `matched_subscription_limit`, `filter_evaluation_limit`, or
|
||||
* `delivery_rate_limit`.
|
||||
* `matched_subscription_limit`, `filter_evaluation_limit`,
|
||||
* `delivery_rate_limit`, or `backlog_overflow` when undelivered events were
|
||||
* shed to stay inside a backlog cap.
|
||||
* @property {number} ts Milliseconds since the epoch.
|
||||
*/
|
||||
|
||||
|
||||
Reference in New Issue
Block a user