Skip to content

feat: implement priority-based queue #350

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Abd-Standard merged 1 commit into from
Jun 30, 2026
Merged

feat: implement priority-based queue #350

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
44 changes: 43 additions & 1 deletion listener/src/services/event-processing-queue.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,11 +2,18 @@ import * as StellarSDK from '@stellar/stellar-sdk';
import { ContractConfig } from '../types';
import logger from '../utils/logger';

export enum Priority {
Low = 0,
Medium = 1,
High = 2,
}

export interface EventProcessingQueueOptions {
maxConcurrency?: number;
pollIntervalMs?: number;
maxRetries?: number;
baseDelayMs?: number;
priorityWeights?: { high: number; medium: number; low: number };
}

export type EventProcessor = (
Expand All @@ -22,13 +29,16 @@ interface QueuedEvent {
retryCount: number;
nextRetryAt: number;
fingerprint: string;
priority: Priority;
enqueuedAt: number;
}

const DEFAULTS = {
maxConcurrency: 1,
pollIntervalMs: 1_000,
maxRetries: 3,
baseDelayMs: 2_000,
priorityWeights: { high: 5, medium: 2, low: 1 },
};

export class EventProcessingQueue {
Expand All @@ -39,21 +49,25 @@ export class EventProcessingQueue {
private readonly pollIntervalMs: number;
private readonly maxRetries: number;
private readonly baseDelayMs: number;
private readonly priorityWeights: { high: number; medium: number; low: number };
private readonly processor: EventProcessor;
private timer: ReturnType<typeof setInterval> | null = null;
private priorityCounters: { high: number; medium: number; low: number } = { high: 0, medium: 0, low: 0 };

constructor(processor: EventProcessor, options?: EventProcessingQueueOptions) {
this.processor = processor;
this.maxConcurrency = Math.max(1, options?.maxConcurrency ?? DEFAULTS.maxConcurrency);
this.pollIntervalMs = options?.pollIntervalMs ?? DEFAULTS.pollIntervalMs;
this.maxRetries = options?.maxRetries ?? DEFAULTS.maxRetries;
this.baseDelayMs = options?.baseDelayMs ?? DEFAULTS.baseDelayMs;
this.priorityWeights = options?.priorityWeights ?? DEFAULTS.priorityWeights;
}

enqueue(
event: StellarSDK.rpc.Api.EventResponse,
contractConfig: ContractConfig,
requestId?: string
requestId?: string,
priority: Priority = Priority.Medium
): boolean {
const fingerprint = buildEventFingerprint(event, contractConfig.address);

Expand All @@ -77,6 +91,7 @@ export class EventProcessingQueue {
delayMs,
nextRetryAt: new Date(nextRetryAt).toISOString(),
maxRetries: this.maxRetries,
priority: Priority[priority],
});

this.queuedFingerprints.add(fingerprint);
Expand All @@ -87,6 +102,8 @@ export class EventProcessingQueue {
retryCount: 0,
nextRetryAt,
fingerprint,
priority,
enqueuedAt: Date.now(),
});

return true;
Expand Down Expand Up @@ -124,10 +141,22 @@ export class EventProcessingQueue {

const due = this.queue
.filter((item) => item.nextRetryAt <= now && !this.activeFingerprints.has(item.fingerprint))
.sort((a, b) => {
const priorityA = this.getWeightedPriority(a);
const priorityB = this.getWeightedPriority(b);
if (priorityB !== priorityA) return priorityB - priorityA;
return a.enqueuedAt - b.enqueuedAt;
})
.slice(0, available);

if (due.length === 0) return;

for (const item of due) {
if (item.priority === Priority.High) this.priorityCounters.high++;
else if (item.priority === Priority.Medium) this.priorityCounters.medium++;
else this.priorityCounters.low++;
}

const selectedFingerprints = new Set(due.map((item) => item.fingerprint));

this.queue = this.queue.filter(
Expand All @@ -148,6 +177,19 @@ export class EventProcessingQueue {
}
}

private getWeightedPriority(item: QueuedEvent): number {
const basePriority = item.priority;
const age = Date.now() - item.enqueuedAt;
const ageBonus = Math.floor(age / 60000);

let weight = 0;
if (item.priority === Priority.High) weight = this.priorityWeights.high;
else if (item.priority === Priority.Medium) weight = this.priorityWeights.medium;
else weight = this.priorityWeights.low;

return basePriority + ageBonus + weight;
}

private async processItem(item: QueuedEvent): Promise<void> {
this.activeFingerprints.add(item.fingerprint);

Expand Down
48 changes: 45 additions & 3 deletions listener/src/services/notification-retry-queue.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,12 +5,19 @@ import { getEventName } from '../utils/event-utils';
import { getNotificationAnalyticsAggregator, NotificationAnalyticsAggregator } from './notification-analytics-aggregator';
import { NotificationType } from '../types/scheduled-notification';

export enum Priority {
Low = 0,
Medium = 1,
High = 2,
}

export interface RetryQueueOptions {
baseDelayMs?: number;
multiplier?: number;
jitter?: boolean;
maxRetries?: number;
processIntervalMs?: number;
priorityWeights?: { high: number; medium: number; low: number };
}

interface RetryItem {
Expand All @@ -19,6 +26,8 @@ interface RetryItem {
retryCount: number;
nextRetryAt: number;
requestId?: string;
priority: Priority;
enqueuedAt: number;
}

const DEFAULTS = {
Expand All @@ -27,6 +36,7 @@ const DEFAULTS = {
jitter: true,
maxRetries: 5,
processIntervalMs: 5_000,
priorityWeights: { high: 5, medium: 2, low: 1 },
};

export type NotificationFn = (
Expand All @@ -43,9 +53,11 @@ export class NotificationRetryQueue {
private readonly jitter: boolean;
private readonly maxRetries: number;
private readonly processIntervalMs: number;
private readonly priorityWeights: { high: number; medium: number; low: number };
private timer: ReturnType<typeof setInterval> | null = null;
private readonly notificationFn: NotificationFn;
private readonly analytics: NotificationAnalyticsAggregator | null;
private priorityCounters: { high: number; medium: number; low: number } = { high: 0, medium: 0, low: 0 };

constructor(notificationFn: NotificationFn, options?: RetryQueueOptions) {
this.notificationFn = notificationFn;
Expand All @@ -54,13 +66,15 @@ export class NotificationRetryQueue {
this.jitter = options?.jitter ?? DEFAULTS.jitter;
this.maxRetries = options?.maxRetries ?? DEFAULTS.maxRetries;
this.processIntervalMs = options?.processIntervalMs ?? DEFAULTS.processIntervalMs;
this.priorityWeights = options?.priorityWeights ?? DEFAULTS.priorityWeights;
this.analytics = getNotificationAnalyticsAggregator();
}

enqueue(
event: StellarSDK.rpc.Api.EventResponse,
contractConfig: ContractConfig,
requestId?: string
requestId?: string,
priority: Priority = Priority.Medium
): void {
const fingerprint = buildRetryFingerprint(event, contractConfig.address);

Expand All @@ -84,10 +98,11 @@ export class NotificationRetryQueue {
delayMs,
nextRetryAt: new Date(nextRetryAt).toISOString(),
maxRetries: this.maxRetries,
priority: Priority[priority],
});

this.queuedFingerprints.add(fingerprint);
this.queue.push({ event, contractConfig, retryCount: 0, nextRetryAt, requestId });
this.queue.push({ event, contractConfig, retryCount: 0, nextRetryAt, requestId, priority, enqueuedAt: Date.now() });
}

start(): void {
Expand All @@ -112,14 +127,41 @@ export class NotificationRetryQueue {

private async processQueue(): Promise<void> {
const now = Date.now();
const due = this.queue.filter((item) => item.nextRetryAt <= now);
const due = this.queue
.filter((item) => item.nextRetryAt <= now)
.sort((a, b) => {
const priorityA = this.getWeightedPriority(a);
const priorityB = this.getWeightedPriority(b);
if (priorityB !== priorityA) return priorityB - priorityA;
return a.enqueuedAt - b.enqueuedAt;
});

this.queue = this.queue.filter((item) => item.nextRetryAt > now);

for (const item of due) {
if (item.priority === Priority.High) this.priorityCounters.high++;
else if (item.priority === Priority.Medium) this.priorityCounters.medium++;
else this.priorityCounters.low++;
}

for (const item of due) {
await this.retryItem(item);
}
}

private getWeightedPriority(item: RetryItem): number {
const basePriority = item.priority;
const age = Date.now() - item.enqueuedAt;
const ageBonus = Math.floor(age / 60000);

let weight = 0;
if (item.priority === Priority.High) weight = this.priorityWeights.high;
else if (item.priority === Priority.Medium) weight = this.priorityWeights.medium;
else weight = this.priorityWeights.low;

return basePriority + ageBonus + weight;
}

private async retryItem(item: RetryItem): Promise<void> {
const attempt = item.retryCount + 1;
const fingerprint = buildRetryFingerprint(item.event, item.contractConfig.address);
Expand Down
Loading