Merge pull request #28 from WaveringAna/codex/fix-unbounded-revalidation-queue-vulnerability
Cap and deduplicate firehose revalidation queue
8 unmodified lines910111213141516171813 unmodified lines8 unmodified lines
# Redis (cache invalidation + revalidation queue)REDIS_URL=redis://localhost:6379WISP_REVALIDATE_FAILURE_BACKOFF_SECONDS=600WISP_REVALIDATE_STREAM_MAXLEN=10000WISP_REVALIDATE_DEDUPE_TTL_SECONDS=60
# S3 Storage (leave empty for local disk fallback)S3_BUCKET=S3_REGION=auto13 unmodified lines26 unmodified lines2728293031323334353615 unmodified lines26 unmodified lines // Redis revalidation queue redisUrl: process.env.REDIS_URL, revalidateStream: process.env.WISP_REVALIDATE_STREAM || 'wisp:revalidate', revalidateGroup: process.env.WISP_REVALIDATE_GROUP || 'firehose-service', revalidateStreamMaxLen: parseInt(process.env.WISP_REVALIDATE_STREAM_MAXLEN || '10000', 10), revalidateDedupeTtlSeconds: parseInt(process.env.WISP_REVALIDATE_DEDUPE_TTL_SECONDS || '60', 10), cacheInvalidationStream: process.env.WISP_CACHE_INVALIDATION_STREAM || 'wisp:cache-invalidate-stream', cacheInvalidationStreamMaxLen: parseInt(process.env.WISP_CACHE_INVALIDATION_STREAM_MAXLEN || '10000', 10),
// Leader election (for distributed HA deployments)15 unmodified lines80 unmodified lines818283848586878889909192939495969798991001012 unmodified lines1041051061071081091101111121131141151168 unmodified lines80 unmodified linesexport async function enqueueSiteRevalidation(did: string, rkey: string, reason: string): Promise<boolean> { const redis = getPublisher() if (!redis) return false
const dedupeKey = `revalidate:site:other:${did}:${rkey}` let dedupeAcquired = false let enqueued = false try { const set = await redis.set(dedupeKey, '1', 'EX', config.revalidateDedupeTtlSeconds, 'NX') if (!set) return false dedupeAcquired = true
const streamId = await redis.xadd( config.revalidateStream, 'MAXLEN', '~', config.revalidateStreamMaxLen.toString(), '*', 'did', did, 'rkey',2 unmodified lines reason, 'ts', Date.now().toString(), ) enqueued = true logger.info(`[Revalidate] Enqueued ${did}/${rkey} after firehose failure`, { reason, streamId }) return true } catch (err) { if (dedupeAcquired && !enqueued) await redis.del(dedupeKey).catch(() => undefined) logger.error('[Revalidate] Failed to enqueue site after firehose failure', err, { did, rkey, reason }) return false }}8 unmodified lines