feat: add durable asset synchronization worker
This commit is contained in:
+222
@@ -0,0 +1,222 @@
|
||||
import { ConvexStore, type RefreshLease } from '../src/lib/server/store';
|
||||
import { TokenCipher } from '../src/lib/server/security';
|
||||
import { EveSso, type VerifiedGrant } from '../src/lib/server/sso';
|
||||
import { EsiClient, EsiError, EsiHttpError, EsiRateLimitError } from '../src/lib/server/esi';
|
||||
|
||||
/** Conservative prototype budgets; these are intentionally configurable and bounded. */
|
||||
export const SYNC_WORKER_DEFAULTS = {
|
||||
pollMs: 1_000,
|
||||
leaseMs: 10 * 60_000,
|
||||
heartbeatMs: 60_000,
|
||||
maxIdlePolls: 0,
|
||||
} as const;
|
||||
|
||||
export type WorkerConfig = {
|
||||
store: ConvexStore;
|
||||
cipher: TokenCipher;
|
||||
sso: EveSso;
|
||||
esi: EsiClient;
|
||||
workerId: string;
|
||||
pollMs: number;
|
||||
leaseMs: number;
|
||||
heartbeatMs: number;
|
||||
sleep: (ms: number) => Promise<void>;
|
||||
};
|
||||
|
||||
function fixedFailure(): string { return 'Asset synchronization failed'; }
|
||||
const DEFAULT_SYNC_COOLDOWN_GROUP = 'esi:assets';
|
||||
function cooldownContext(userId: string, characterId: number): string { return `character:${userId}:${characterId}`; }
|
||||
|
||||
export class MutationUncertainError extends Error {
|
||||
constructor(cause: unknown) { super('Synchronization mutation acknowledgement was uncertain'); this.name = 'MutationUncertainError'; this.cause = cause; }
|
||||
}
|
||||
|
||||
function permanentMutationFailure(error: unknown): boolean {
|
||||
const message = error instanceof Error ? error.message : String(error);
|
||||
return /lease is not valid|authorization (?:is )?stale|authorization changed|generation is not current|already leased|Invalid asset chunk|Asset character mismatch|chunks must be staged|chunk already exists|Snapshot is not staging|Snapshot is incomplete|Invalid snapshot|not active|missing|Invalid refreshed scopes/.test(message);
|
||||
}
|
||||
|
||||
async function retryMutation<T>(operation: () => Promise<T>, renew?: () => Promise<boolean>): Promise<T> {
|
||||
let lastError: unknown;
|
||||
for (let attempt = 0; attempt < 2; attempt += 1) {
|
||||
try { return await operation(); }
|
||||
catch (error) {
|
||||
lastError = error;
|
||||
if (permanentMutationFailure(error)) throw error;
|
||||
if (attempt === 0 && renew) {
|
||||
try { if (!(await renew())) throw new Error('Synchronization lease is not valid'); }
|
||||
catch (renewError) {
|
||||
if (permanentMutationFailure(renewError)) throw renewError;
|
||||
lastError = renewError;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
throw new MutationUncertainError(lastError);
|
||||
}
|
||||
|
||||
/** Process one at-least-once delivery. Every write is fenced by the durable lease. */
|
||||
export async function processSyncJob(config: WorkerConfig): Promise<boolean> {
|
||||
const claimed = await config.store.claimSyncJob(config.workerId, config.leaseMs);
|
||||
if (!claimed) return false;
|
||||
let lostLease = false;
|
||||
let refreshLease: RefreshLease | null = null;
|
||||
let refreshAttempted = false;
|
||||
const heartbeat = setInterval(() => {
|
||||
void config.store.renewSyncJob(claimed.jobId, claimed.leaseToken, config.workerId, config.leaseMs).then((ok) => {
|
||||
if (!ok) lostLease = true;
|
||||
}).catch(() => { lostLease = true; });
|
||||
}, config.heartbeatMs);
|
||||
const stopHeartbeat = () => clearInterval(heartbeat);
|
||||
try {
|
||||
const refresh = await config.store.acquireRefreshLease(claimed.userId, claimed.characterId, config.workerId);
|
||||
refreshLease = refresh;
|
||||
if (!refresh) {
|
||||
await config.store.deferSyncJob(claimed.jobId, claimed.leaseToken, Date.now() + 1_000);
|
||||
return true;
|
||||
}
|
||||
|
||||
let previous: VerifiedGrant;
|
||||
try {
|
||||
previous = await config.cipher.open<VerifiedGrant>(refresh.encryptedTokens, `character:${claimed.characterId}`);
|
||||
if (previous.characterId !== claimed.characterId) throw new Error('authorization mismatch');
|
||||
} catch {
|
||||
await config.store.failSyncJob(claimed.jobId, claimed.leaseToken).catch(() => {});
|
||||
return true;
|
||||
}
|
||||
|
||||
const needsRefresh = refresh.tokenExpiresAt <= Date.now() + 60_000;
|
||||
let grant: VerifiedGrant = previous;
|
||||
let authorityVersion = { grantRevision: refresh.grantRevision, authorizationEpoch: refresh.authorizationEpoch, credentialVersion: refresh.credentialVersion, scopes: refresh.scopes };
|
||||
if (!needsRefresh) {
|
||||
// A lease acquired for a still-valid token is only a serialization guard;
|
||||
// release it before paging so another job is not blocked for two minutes.
|
||||
await retryMutation(() => config.store.releaseRefreshLease
|
||||
? config.store.releaseRefreshLease(claimed.userId, claimed.characterId, refresh.leaseToken, config.workerId)
|
||||
: Promise.resolve(true));
|
||||
refreshLease = null;
|
||||
}
|
||||
if (needsRefresh) {
|
||||
try {
|
||||
const dispatched = await config.store.markRefreshDispatched(claimed.userId, claimed.characterId, refresh.leaseToken, config.workerId);
|
||||
if (!dispatched) {
|
||||
await config.store.failSyncJob(claimed.jobId, claimed.leaseToken).catch(() => {});
|
||||
return true;
|
||||
}
|
||||
refreshAttempted = true;
|
||||
// Exactly one remote refresh attempt. An error is treated as uncertain:
|
||||
// rotating refresh tokens cannot safely be replayed after a lost result.
|
||||
grant = await config.sso.refresh(previous.refreshToken);
|
||||
} catch {
|
||||
await config.store.markRefreshUncertain(claimed.userId, claimed.characterId, refresh.leaseToken, config.workerId, refresh.expectedVersion).catch(() => {});
|
||||
return true;
|
||||
}
|
||||
if (grant.characterId !== claimed.characterId || grant.ownerHash !== previous.ownerHash) {
|
||||
await config.store.markRefreshUncertain(claimed.userId, claimed.characterId, refresh.leaseToken, config.workerId, refresh.expectedVersion).catch(() => {});
|
||||
return true;
|
||||
}
|
||||
const committed = await config.store.commitRefresh(
|
||||
claimed.userId,
|
||||
claimed.characterId,
|
||||
refresh.leaseToken,
|
||||
config.workerId,
|
||||
refresh.expectedVersion,
|
||||
{ encryptedTokens: await config.cipher.seal(grant, `character:${claimed.characterId}`), tokenExpiresAt: grant.expiresAt, scopes: grant.scopes, jobId: claimed.jobId, jobLeaseToken: claimed.leaseToken },
|
||||
);
|
||||
if (!committed) {
|
||||
await config.store.markRefreshUncertain(claimed.userId, claimed.characterId, refresh.leaseToken, config.workerId, refresh.expectedVersion).catch(() => {});
|
||||
await config.store.failSyncJob(claimed.jobId, claimed.leaseToken).catch(() => {});
|
||||
return true;
|
||||
}
|
||||
authorityVersion = committed;
|
||||
// A refreshed credential cannot publish into a generation fenced to the
|
||||
// old credential version. Discard only that staging generation.
|
||||
if (claimed.generation !== undefined) await retryMutation(
|
||||
() => config.store.restartSyncJobGeneration(claimed.jobId, claimed.leaseToken),
|
||||
() => config.store.renewSyncJob(claimed.jobId, claimed.leaseToken, config.workerId, config.leaseMs),
|
||||
);
|
||||
}
|
||||
const currentAuthority = await retryMutation(
|
||||
() => config.store.refreshSyncJobAuthority(claimed.jobId, claimed.leaseToken, authorityVersion),
|
||||
() => config.store.renewSyncJob(claimed.jobId, claimed.leaseToken, config.workerId, config.leaseMs),
|
||||
);
|
||||
if (!grant.scopes.includes('esi-assets.read_assets.v1') || !currentAuthority.scopes.includes('esi-assets.read_assets.v1') || lostLease) {
|
||||
await config.store.failSyncJob(claimed.jobId, claimed.leaseToken).catch(() => {});
|
||||
return true;
|
||||
}
|
||||
authorityVersion = currentAuthority;
|
||||
const generation = await retryMutation(
|
||||
() => config.store.beginSyncJob(claimed.jobId, claimed.leaseToken),
|
||||
() => config.store.renewSyncJob(claimed.jobId, claimed.leaseToken, config.workerId, config.leaseMs),
|
||||
);
|
||||
|
||||
const context = cooldownContext(claimed.userId, claimed.characterId);
|
||||
const cooldown = await config.store.cooldown(context, claimed.cooldownGroup);
|
||||
if (cooldown && cooldown.until > Date.now()) await config.sleep(cooldown.until - Date.now());
|
||||
const authority = {
|
||||
userId: claimed.userId,
|
||||
principal: grant.ownerHash,
|
||||
characterId: claimed.characterId,
|
||||
scopes: authorityVersion.scopes,
|
||||
grantRevision: authorityVersion.grantRevision,
|
||||
authorizationEpoch: authorityVersion.authorizationEpoch,
|
||||
credentialVersion: authorityVersion.credentialVersion,
|
||||
} as const;
|
||||
const snapshot = await config.esi.getAssets(claimed.characterId, grant.accessToken, authority);
|
||||
const chunks = Math.ceil(snapshot.assets.length / 250);
|
||||
for (let index = 0; index < chunks; index += 1) {
|
||||
if (lostLease) throw new Error('lease lost');
|
||||
await retryMutation(
|
||||
() => config.store.stageSyncAssets(claimed.jobId, claimed.leaseToken, index, snapshot.assets.slice(index * 250, (index + 1) * 250)),
|
||||
() => config.store.renewSyncJob(claimed.jobId, claimed.leaseToken, config.workerId, config.leaseMs),
|
||||
);
|
||||
}
|
||||
await retryMutation(
|
||||
() => config.store.commitSyncJob(claimed.jobId, claimed.leaseToken, chunks, snapshot.syncedAt, snapshot.expiresAt),
|
||||
() => config.store.renewSyncJob(claimed.jobId, claimed.leaseToken, config.workerId, config.leaseMs),
|
||||
);
|
||||
return true;
|
||||
} catch (error) {
|
||||
if (refreshAttempted && refreshLease) await config.store.markRefreshUncertain(claimed.userId, claimed.characterId, refreshLease.leaseToken, config.workerId, refreshLease.expectedVersion).catch(() => {});
|
||||
if (error instanceof MutationUncertainError) {
|
||||
await config.store.deferSyncJob(claimed.jobId, claimed.leaseToken, Date.now() + 1_000).catch(() => {});
|
||||
} else if (error instanceof EsiRateLimitError) {
|
||||
const retryAt = Math.min(Date.now() + 24 * 60 * 60_000, Math.max(Date.now() + 1_000, error.retryAt));
|
||||
await config.store.deferSyncJob(claimed.jobId, claimed.leaseToken, retryAt, { context: cooldownContext(claimed.userId, claimed.characterId), until: retryAt, group: error.cooldown.group ?? DEFAULT_SYNC_COOLDOWN_GROUP }).catch(() => {});
|
||||
} else if (error instanceof EsiError && (error.code === 'network' || error.code === 'deadline' || (error instanceof EsiHttpError && (error.status === 408 || error.status === 420 || error.status === 425 || error.status === 429 || error.status >= 500)))) {
|
||||
const retryAt = Math.min(Date.now() + 24 * 60 * 60_000, Date.now() + 1_000 * 2 ** Math.max(0, claimed.attempts - 1));
|
||||
await config.store.deferSyncJob(claimed.jobId, claimed.leaseToken, retryAt).catch(() => {});
|
||||
} else {
|
||||
await config.store.failSyncJob(claimed.jobId, claimed.leaseToken).catch(() => {});
|
||||
}
|
||||
return true;
|
||||
} finally {
|
||||
stopHeartbeat();
|
||||
}
|
||||
}
|
||||
|
||||
function env(name: string): string {
|
||||
const value = process.env[name];
|
||||
if (!value) throw new Error(`Sync worker configuration missing: ${name}`);
|
||||
return value;
|
||||
}
|
||||
|
||||
export async function runSyncWorker(): Promise<void> {
|
||||
const workerId = process.env.SYNC_WORKER_ID ?? `sync-worker-${crypto.randomUUID()}`;
|
||||
const pollMs = Math.max(250, Math.min(30_000, Number(process.env.SYNC_WORKER_POLL_MS ?? SYNC_WORKER_DEFAULTS.pollMs)));
|
||||
const leaseMs = Math.max(60_000, Math.min(60 * 60_000, Number(process.env.SYNC_WORKER_LEASE_MS ?? SYNC_WORKER_DEFAULTS.leaseMs)));
|
||||
const heartbeatMs = Math.max(10_000, Math.min(10 * 60_000, Number(process.env.SYNC_WORKER_HEARTBEAT_MS ?? SYNC_WORKER_DEFAULTS.heartbeatMs)));
|
||||
const store = new ConvexStore(env('CONVEX_URL'), undefined, env('CONVEX_SYNC_WORKER_SECRET'));
|
||||
const cipher = new TokenCipher(env('TOKEN_ENCRYPTION_KEY'));
|
||||
const userAgent = env('ESI_USER_AGENT');
|
||||
const sso = new EveSso({ clientId: env('EVE_CLIENT_ID'), clientSecret: process.env.EVE_CLIENT_SECRET || undefined, redirectUri: process.env.EVE_CALLBACK_URL ?? 'http://localhost/unused-worker-callback', userAgent });
|
||||
const esi = new EsiClient({ userAgent });
|
||||
const config: WorkerConfig = { store, cipher, sso, esi, workerId, pollMs, leaseMs, heartbeatMs, sleep: (ms) => new Promise((resolve) => setTimeout(resolve, ms)) };
|
||||
if (process.env.SYNC_WORKER_STARTUP_CHECK === '1') return;
|
||||
for (;;) {
|
||||
const handled = await processSyncJob(config);
|
||||
if (!handled) await config.sleep(pollMs);
|
||||
}
|
||||
}
|
||||
|
||||
if (import.meta.main) await runSyncWorker();
|
||||
Reference in New Issue
Block a user