@@ -3,89 +3,131 @@ import assert from "node:assert/strict"
33export interface SerializationEvent {
44 event : "start" | "finish"
55 messageId : string
6+ attempt : number
7+ processId : number
68 at : number
79}
810
911export interface SerializationProof {
1012 executions : number
1113 retried : boolean
12- overlapped : boolean
14+ supersededOverlap : boolean
15+ }
16+
17+ interface Execution {
18+ messageId : string
19+ attempt : number
20+ processId : number
21+ startedAt : number
22+ finishedAt : number
1323}
1424
1525export function parseSerializationEvent ( line : string ) : SerializationEvent {
1626 const event = JSON . parse ( line ) as Partial < SerializationEvent >
1727 if (
1828 ( event . event !== "start" && event . event !== "finish" ) ||
1929 typeof event . messageId !== "string" ||
30+ typeof event . attempt !== "number" ||
31+ typeof event . processId !== "number" ||
2032 typeof event . at !== "number"
2133 ) {
2234 throw new TypeError ( "invalid serialization event" )
2335 }
24- return { event : event . event , messageId : event . messageId , at : event . at }
36+ return {
37+ event : event . event ,
38+ messageId : event . messageId ,
39+ attempt : event . attempt ,
40+ processId : event . processId ,
41+ at : event . at ,
42+ }
2543}
2644
2745// One identity commits one state transition at a time. The control file is
2846// written outside the transaction, so it records execution attempts rather than
29- // commits: a worker that loses its lease keeps running until it notices, and
30- // its replacement executes the same message again. A superseded attempt may
31- // therefore overlap anything, because its write is fenced out and the committed
32- // state is what proves it.
47+ // commits: a worker that loses its lease keeps running until it notices, and its
48+ // replacement executes the same message under a higher attempt. The superseded
49+ // attempt may therefore overlap anything, because its write is fenced out and
50+ // the committed state is what proves it.
3351//
34- // The surviving attempts are the ones that count, so the last attempt of each
35- // message must still have the identity to itself.
52+ // Each event carries its attempt and process, so a start pairs with its own
53+ // finish rather than with whichever finish arrived next. Without that, a
54+ // superseded attempt finishing late reads as its replacement finishing, and a
55+ // second message could then overlap a replacement that is still running.
3656export function assertSerializedExecution (
3757 events : readonly SerializationEvent [ ] ,
3858 options : { messageCount : number } ,
3959) : SerializationProof {
40- const ordered = [ ...events ] . sort ( ( left , right ) => left . at - right . at )
60+ const executions = pairExecutions ( events )
61+
62+ const messageIds = new Set ( executions . map ( ( execution ) => execution . messageId ) )
63+ assert . equal (
64+ messageIds . size ,
65+ options . messageCount ,
66+ `expected ${ options . messageCount } messages to run, saw ${ messageIds . size } ` ,
67+ )
68+
69+ const survivingAttempt = new Map < string , number > ( )
70+ for ( const execution of executions ) {
71+ const highest = survivingAttempt . get ( execution . messageId ) ?? 0
72+ if ( execution . attempt > highest ) survivingAttempt . set ( execution . messageId , execution . attempt )
73+ }
74+ const surviving = executions . filter (
75+ ( execution ) => survivingAttempt . get ( execution . messageId ) === execution . attempt ,
76+ )
4177
42- const lastStartedAt = new Map < string , number > ( )
43- for ( const event of ordered ) {
44- if ( event . event === "start" ) lastStartedAt . set ( event . messageId , event . at )
78+ for ( const [ index , execution ] of surviving . entries ( ) ) {
79+ for ( const other of surviving . slice ( index + 1 ) ) {
80+ assert (
81+ ! overlaps ( execution , other ) ,
82+ `${ describe ( execution ) } and ${ describe ( other ) } overlap, and neither was superseded` ,
83+ )
84+ }
4585 }
4686
47- const open = new Map < string , number [ ] > ( )
48- const surviving = new Set < string > ( )
49- let running = 0
50- let concurrent = 0
51- let executions = 0
87+ const supersededOverlap = executions . some ( ( execution ) =>
88+ executions . some ( ( other ) => other !== execution && overlaps ( execution , other ) ) ,
89+ )
5290
53- for ( const event of ordered ) {
91+ return {
92+ executions : executions . length ,
93+ retried : executions . length > options . messageCount ,
94+ supersededOverlap,
95+ }
96+ }
97+
98+ function pairExecutions ( events : readonly SerializationEvent [ ] ) : Execution [ ] {
99+ const started = new Map < string , SerializationEvent > ( )
100+ const executions : Execution [ ] = [ ]
101+
102+ for ( const event of [ ...events ] . sort ( ( left , right ) => left . at - right . at ) ) {
103+ const key = `${ event . messageId } #${ event . attempt } #${ event . processId } `
54104 if ( event . event === "start" ) {
55- open . set ( event . messageId , [ ...( open . get ( event . messageId ) ?? [ ] ) , event . at ] )
56- executions += 1
57- running += 1
58- concurrent = Math . max ( concurrent , running )
59- if ( event . at === lastStartedAt . get ( event . messageId ) ) {
60- surviving . add ( event . messageId )
61- assert . equal (
62- surviving . size ,
63- 1 ,
64- `${ [ ...surviving ] . join ( " and " ) } overlap, and neither was superseded by a retry` ,
65- )
66- }
105+ assert ( ! started . has ( key ) , `${ describe ( event ) } started twice` )
106+ started . set ( key , event )
67107 continue
68108 }
69- const openForMessage = open . get ( event . messageId ) ?? [ ]
70- assert ( openForMessage . length > 0 , `finish of ${ event . messageId } has no matching start` )
71- const startedAt = openForMessage . pop ( )
72- open . set ( event . messageId , openForMessage )
73- running -= 1
74- if ( startedAt === lastStartedAt . get ( event . messageId ) ) surviving . delete ( event . messageId )
109+ const start = started . get ( key )
110+ assert ( start !== undefined , `${ describe ( event ) } finished with no matching start` )
111+ started . delete ( key )
112+ executions . push ( {
113+ messageId : event . messageId ,
114+ attempt : event . attempt ,
115+ processId : event . processId ,
116+ startedAt : start . at ,
117+ finishedAt : event . at ,
118+ } )
75119 }
76120
77- assert . equal ( running , 0 , "an execution never wrote a finish" )
121+ const unfinished = [ ...started . values ( ) ] . map ( describe )
122+ assert . equal ( unfinished . length , 0 , `${ unfinished . join ( ", " ) } never wrote a finish` )
78123
79- const messageIds = new Set ( ordered . map ( ( event ) => event . messageId ) )
80- assert . equal (
81- messageIds . size ,
82- options . messageCount ,
83- `expected ${ options . messageCount } messages to run, saw ${ messageIds . size } ` ,
84- )
124+ return executions
125+ }
85126
86- return {
87- executions,
88- retried : executions > options . messageCount ,
89- overlapped : concurrent > 1 ,
90- }
127+ function overlaps ( left : Execution , right : Execution ) : boolean {
128+ return left . startedAt < right . finishedAt && right . startedAt < left . finishedAt
129+ }
130+
131+ function describe ( execution : { messageId : string ; attempt : number } ) : string {
132+ return `${ execution . messageId } attempt ${ execution . attempt } `
91133}
0 commit comments