66 calculateEffectiveScheduleTime ,
77 calculateNextNominalTimestamp ,
88 calculateSchedulePhase ,
9+ SCHEDULE_PHASE_DENOMINATOR ,
910 ScheduleEngine ,
1011} from "../src/index.js" ;
1112import { calculateDistributedExecutionTime } from "../src/engine/distributedScheduling.js" ;
@@ -27,7 +28,7 @@ describe("ScheduleEngine Integration (part 2)", () => {
2728 redis : redisOptions ,
2829 distributionWindow : { seconds : 10 } ,
2930 schedulePhaseSecret : "test-schedule-phase-secret" ,
30- cronSpreadEnabled : false ,
31+ cronSpreadFraction : 0 ,
3132 worker : {
3233 concurrency : 1 ,
3334 disabled : true , // Don't actually run the worker — calling triggerScheduledTask directly
@@ -169,7 +170,7 @@ describe("ScheduleEngine Integration (part 2)", () => {
169170 redis : redisOptions ,
170171 distributionWindow : { seconds : 10 } ,
171172 schedulePhaseSecret,
172- cronSpreadEnabled : true ,
173+ cronSpreadFraction : 1 ,
173174 worker : {
174175 concurrency : 1 ,
175176 disabled : true ,
@@ -366,4 +367,126 @@ describe("ScheduleEngine Integration (part 2)", () => {
366367 }
367368 }
368369 ) ;
370+
371+ containerTest (
372+ "gates cron spread per schedule via the rollout fraction" ,
373+ { timeout : 30_000 } ,
374+ async ( { prisma, redisOptions } ) => {
375+ const schedulePhaseSecret = "test-schedule-phase-secret" ;
376+
377+ const organization = await prisma . organization . create ( {
378+ data : { title : "Spread Fraction Org" , slug : "spread-fraction-org" } ,
379+ } ) ;
380+ const project = await prisma . project . create ( {
381+ data : {
382+ name : "Spread Fraction Project" ,
383+ slug : "spread-fraction-project" ,
384+ externalRef : "spread-fraction-ref" ,
385+ organizationId : organization . id ,
386+ } ,
387+ } ) ;
388+ const environment = await prisma . runtimeEnvironment . create ( {
389+ data : {
390+ slug : "spread-fraction-env" ,
391+ type : "PRODUCTION" ,
392+ projectId : project . id ,
393+ organizationId : organization . id ,
394+ apiKey : "tr_spread_fraction" ,
395+ pkApiKey : "pk_spread_fraction" ,
396+ shortcode : "spread" ,
397+ } ,
398+ } ) ;
399+ const taskSchedule = await prisma . taskSchedule . create ( {
400+ data : {
401+ friendlyId : "sched_spread_fraction" ,
402+ taskIdentifier : "spread-fraction-task" ,
403+ projectId : project . id ,
404+ deduplicationKey : "spread-fraction-dedup" ,
405+ generatorExpression : "*/5 * * * *" ,
406+ generatorDescription : "Every 5 minutes" ,
407+ timezone : "UTC" ,
408+ type : "DECLARATIVE" ,
409+ } ,
410+ } ) ;
411+ const scheduleInstance = await prisma . taskScheduleInstance . create ( {
412+ data : {
413+ taskScheduleId : taskSchedule . id ,
414+ environmentId : environment . id ,
415+ projectId : project . id ,
416+ } ,
417+ } ) ;
418+
419+ const phase = calculateSchedulePhase ( {
420+ secret : schedulePhaseSecret ,
421+ environmentId : environment . id ,
422+ deduplicationKey : taskSchedule . deduplicationKey ,
423+ } ) ;
424+
425+ // The gate is `phase < fraction * DENOMINATOR`. Dividing and multiplying
426+ // by 2^31 is exact in floating point, so `phase / DENOMINATOR` excludes
427+ // this schedule and `(phase + 1) / DENOMINATOR` includes it.
428+ const excludingFraction = phase / SCHEDULE_PHASE_DENOMINATOR ;
429+ const includingFraction = ( phase + 1 ) / SCHEDULE_PHASE_DENOMINATOR ;
430+
431+ const createEngine = ( cronSpreadFraction : number ) =>
432+ new ScheduleEngine ( {
433+ prisma,
434+ redis : redisOptions ,
435+ distributionWindow : { seconds : 10 } ,
436+ schedulePhaseSecret,
437+ cronSpreadFraction,
438+ worker : {
439+ concurrency : 1 ,
440+ disabled : true ,
441+ pollIntervalMs : 1000 ,
442+ } ,
443+ tracer : trace . getTracer ( "test" , "0.0.0" ) ,
444+ onTriggerScheduledTask : async ( ) => ( { success : true } ) ,
445+ isDevEnvironmentConnectedHandler : vi . fn ( ) . mockResolvedValue ( true ) ,
446+ } ) ;
447+
448+ const jobId = `scheduled-task-instance:${ scheduleInstance . id } ;
449+
450+ const excludedEngine = createEngine ( excludingFraction ) ;
451+ try {
452+ await excludedEngine . registerNextTaskScheduleInstance ( { instanceId : scheduleInstance . id } ) ;
453+ const job = await excludedEngine . getJob ( jobId ) ;
454+ const payload = job ! . item as unknown as {
455+ exactScheduleTime : string ;
456+ effectiveScheduleTime : string ;
457+ } ;
458+ // Spread inactive: the effective time is the nominal tick.
459+ expect ( new Date ( payload . effectiveScheduleTime ) ) . toEqual (
460+ new Date ( payload . exactScheduleTime )
461+ ) ;
462+ } finally {
463+ await excludedEngine . quit ( ) ;
464+ }
465+
466+ const includedEngine = createEngine ( includingFraction ) ;
467+ try {
468+ await includedEngine . registerNextTaskScheduleInstance ( { instanceId : scheduleInstance . id } ) ;
469+ const job = await includedEngine . getJob ( jobId ) ;
470+ const payload = job ! . item as unknown as {
471+ exactScheduleTime : string ;
472+ effectiveScheduleTime : string ;
473+ } ;
474+ const nominalAt = new Date ( payload . exactScheduleTime ) ;
475+ const nextNominalAt = calculateNextNominalTimestamp (
476+ taskSchedule . generatorExpression ,
477+ taskSchedule . timezone ,
478+ nominalAt
479+ ) ;
480+ // Spread active with no window configured: the 60s baseline applies.
481+ const { effectiveAt } = calculateEffectiveScheduleTime ( {
482+ nominalAt,
483+ nextNominalAt,
484+ schedulePhase : phase ,
485+ } ) ;
486+ expect ( new Date ( payload . effectiveScheduleTime ) ) . toEqual ( effectiveAt ) ;
487+ } finally {
488+ await includedEngine . quit ( ) ;
489+ }
490+ }
491+ ) ;
369492} ) ;
0 commit comments