diff --git a/api/submitqueue/gateway/proto/gateway.proto b/api/submitqueue/gateway/proto/gateway.proto index 2a14f359..272c7d01 100644 --- a/api/submitqueue/gateway/proto/gateway.proto +++ b/api/submitqueue/gateway/proto/gateway.proto @@ -170,12 +170,18 @@ message GetRequestHistoryByIDRequest { message HistoryEvent { // Time the request-log entry was created, in Unix milliseconds. int64 timestamp_ms = 1; - // Customer-friendly request status recorded by the event. + // Customer-friendly request status recorded by the event. Set only when type is "status". string status = 2; // Error message associated with the event. Empty when absent. string last_error = 3; // Display and debugging metadata associated with this event. Each lifecycle event carries its own values. map metadata = 4; + // What this entry records: "status" when the request reached a position in the pipeline, + // "event" when something happened while it sat at one. Exactly one of status and event is set. + string type = 5; + // Occurrence recorded by the entry, e.g. a build starting or finishing. Set only when type is "event". + // A request records many of these — one per build — and they never change its current status. + string event = 6; } // GetRequestHistoryByIDResponse contains all retained events for one request. diff --git a/api/submitqueue/gateway/protopb/gateway.pb.go b/api/submitqueue/gateway/protopb/gateway.pb.go index ff5e32a1..301a5393 100644 --- a/api/submitqueue/gateway/protopb/gateway.pb.go +++ b/api/submitqueue/gateway/protopb/gateway.pb.go @@ -881,12 +881,18 @@ type HistoryEvent struct { state protoimpl.MessageState `protogen:"open.v1"` // Time the request-log entry was created, in Unix milliseconds. TimestampMs int64 `protobuf:"varint,1,opt,name=timestamp_ms,json=timestampMs,proto3" json:"timestamp_ms,omitempty"` - // Customer-friendly request status recorded by the event. + // Customer-friendly request status recorded by the event. Set only when type is "status". Status string `protobuf:"bytes,2,opt,name=status,proto3" json:"status,omitempty"` // Error message associated with the event. Empty when absent. LastError string `protobuf:"bytes,3,opt,name=last_error,json=lastError,proto3" json:"last_error,omitempty"` // Display and debugging metadata associated with this event. Each lifecycle event carries its own values. - Metadata map[string]string `protobuf:"bytes,4,rep,name=metadata,proto3" json:"metadata,omitempty" protobuf_key:"bytes,1,opt,name=key" protobuf_val:"bytes,2,opt,name=value"` + Metadata map[string]string `protobuf:"bytes,4,rep,name=metadata,proto3" json:"metadata,omitempty" protobuf_key:"bytes,1,opt,name=key" protobuf_val:"bytes,2,opt,name=value"` + // What this entry records: "status" when the request reached a position in the pipeline, + // "event" when something happened while it sat at one. Exactly one of status and event is set. + Type string `protobuf:"bytes,5,opt,name=type,proto3" json:"type,omitempty"` + // Occurrence recorded by the entry, e.g. a build starting or finishing. Set only when type is "event". + // A request records many of these — one per build — and they never change its current status. + Event string `protobuf:"bytes,6,opt,name=event,proto3" json:"event,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -949,6 +955,20 @@ func (x *HistoryEvent) GetMetadata() map[string]string { return nil } +func (x *HistoryEvent) GetType() string { + if x != nil { + return x.Type + } + return "" +} + +func (x *HistoryEvent) GetEvent() string { + if x != nil { + return x.Event + } + return "" +} + // GetRequestHistoryByIDResponse contains all retained events for one request. type GetRequestHistoryByIDResponse struct { state protoimpl.MessageState `protogen:"open.v1"` @@ -1376,13 +1396,15 @@ const file_gateway_proto_rawDesc = "" + "\x0fnext_page_token\x18\x02 \x01(\tR\rnextPageToken\"H\n" + "\x1cGetRequestHistoryByIDRequest\x12\x12\n" + "\x04sqid\x18\x01 \x01(\tR\x04sqid\x12\x14\n" + - "\x05queue\x18\x02 \x01(\tR\x05queue\"\xf7\x01\n" + + "\x05queue\x18\x02 \x01(\tR\x05queue\"\xa1\x02\n" + "\fHistoryEvent\x12!\n" + "\ftimestamp_ms\x18\x01 \x01(\x03R\vtimestampMs\x12\x16\n" + "\x06status\x18\x02 \x01(\tR\x06status\x12\x1d\n" + "\n" + "last_error\x18\x03 \x01(\tR\tlastError\x12P\n" + - "\bmetadata\x18\x04 \x03(\v24.uber.submitqueue.gateway.HistoryEvent.MetadataEntryR\bmetadata\x1a;\n" + + "\bmetadata\x18\x04 \x03(\v24.uber.submitqueue.gateway.HistoryEvent.MetadataEntryR\bmetadata\x12\x12\n" + + "\x04type\x18\x05 \x01(\tR\x04type\x12\x14\n" + + "\x05event\x18\x06 \x01(\tR\x05event\x1a;\n" + "\rMetadataEntry\x12\x10\n" + "\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n" + "\x05value\x18\x02 \x01(\tR\x05value:\x028\x01\"_\n" + diff --git a/api/submitqueue/gateway/protopb/gateway.pb.yarpc.go b/api/submitqueue/gateway/protopb/gateway.pb.yarpc.go index 68c53c35..889e83b7 100644 --- a/api/submitqueue/gateway/protopb/gateway.pb.yarpc.go +++ b/api/submitqueue/gateway/protopb/gateway.pb.yarpc.go @@ -576,75 +576,76 @@ var ( var yarpcFileDescriptorClosuref1a937782ebbded5 = [][]byte{ // gateway.proto []byte{ - 0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xb4, 0x57, 0x4b, 0x6f, 0xdb, 0x46, + 0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xac, 0x57, 0x4b, 0x6f, 0xdb, 0x46, 0x10, 0x0e, 0x45, 0x5b, 0xb6, 0x46, 0xb2, 0x6b, 0x6c, 0xd3, 0x94, 0x50, 0x6d, 0xc4, 0x66, 0x53, 0x47, 0x87, 0x42, 0x0a, 0xdc, 0xf4, 0x81, 0x06, 0x09, 0x10, 0x27, 0xce, 0x03, 0xa8, 0x53, 0x85, 0xae, 0x51, 0x20, 0x40, 0x21, 0xac, 0xa4, 0x89, 0x4c, 0xd8, 0x24, 0xe5, 0xdd, 0xa5, 0x5b, 0x19, - 0x3d, 0xf4, 0xd2, 0x4b, 0xcf, 0xbd, 0xf7, 0xc7, 0xf4, 0xd4, 0xff, 0xd0, 0xff, 0xd1, 0x6b, 0xb1, - 0x0f, 0x52, 0xa4, 0x2d, 0x32, 0x8a, 0x9b, 0x9e, 0xc4, 0xdd, 0x9d, 0xe7, 0xf7, 0xcd, 0xce, 0x8e, - 0x60, 0x65, 0x44, 0x05, 0xfe, 0x48, 0x27, 0xed, 0x31, 0x8b, 0x44, 0x44, 0x9c, 0xb8, 0x8f, 0xac, - 0xcd, 0xe3, 0x7e, 0xe0, 0x8b, 0xd3, 0x18, 0x63, 0x6c, 0x9b, 0xf3, 0xa6, 0x4b, 0xc7, 0x7e, 0xa7, - 0x4f, 0x39, 0x76, 0x06, 0x47, 0x34, 0x1c, 0x61, 0x47, 0x29, 0x98, 0x85, 0xd6, 0x6e, 0xde, 0x49, - 0x65, 0x02, 0x64, 0x23, 0xe4, 0x82, 0x51, 0x81, 0xa3, 0x89, 0x11, 0xcd, 0xed, 0x69, 0x0d, 0xf7, - 0x36, 0xd4, 0xbb, 0x7e, 0x38, 0xf2, 0xf0, 0x34, 0x46, 0x2e, 0x88, 0x03, 0x4b, 0x01, 0x72, 0x4e, - 0x47, 0xe8, 0x58, 0x9b, 0x56, 0xab, 0xe6, 0x25, 0x4b, 0xf7, 0x57, 0x0b, 0x1a, 0x5a, 0x92, 0x8f, - 0xa3, 0x90, 0x63, 0xb1, 0x28, 0xd9, 0x82, 0x06, 0x47, 0x76, 0xe6, 0x0f, 0xb0, 0x17, 0xd2, 0x00, - 0x9d, 0x8a, 0x3a, 0xae, 0x9b, 0xbd, 0x17, 0x34, 0x40, 0xb2, 0x0e, 0x35, 0xe1, 0x07, 0xc8, 0x05, - 0x0d, 0xc6, 0x8e, 0xbd, 0x69, 0xb5, 0x6c, 0x6f, 0xba, 0x41, 0x9a, 0xb0, 0x7c, 0x14, 0x71, 0xa1, - 0x94, 0x17, 0x94, 0x72, 0xba, 0x76, 0x7f, 0xb7, 0xa0, 0xfe, 0x0d, 0x0d, 0x87, 0x49, 0xc4, 0xd7, - 0x61, 0x51, 0xe1, 0x64, 0x82, 0xd0, 0x0b, 0x72, 0x07, 0xaa, 0x1a, 0x18, 0xe5, 0xbc, 0xbe, 0xe3, - 0xb4, 0x15, 0xae, 0x12, 0x9a, 0xb6, 0x41, 0xec, 0x91, 0xfa, 0xf1, 0x8c, 0x1c, 0xb9, 0x0f, 0xcb, - 0x09, 0x34, 0xca, 0xe7, 0xea, 0xce, 0x56, 0x46, 0x27, 0x0f, 0xdd, 0x81, 0xf9, 0xf0, 0x52, 0x15, - 0xd7, 0x85, 0x86, 0x8e, 0xca, 0xa0, 0x43, 0x60, 0x81, 0x9f, 0xfa, 0x43, 0x13, 0x95, 0xfa, 0x76, - 0x5f, 0xc2, 0xca, 0x23, 0x1a, 0x0e, 0xf0, 0x24, 0x89, 0x7d, 0x86, 0x10, 0xb9, 0x01, 0x55, 0x86, - 0x94, 0x47, 0xa1, 0x81, 0xcd, 0xac, 0xa6, 0x79, 0xda, 0x99, 0x3c, 0xdd, 0x35, 0x58, 0x4d, 0x4c, - 0x6a, 0xc7, 0xee, 0x5f, 0x15, 0x58, 0x35, 0xf6, 0x0f, 0xe2, 0x20, 0xa0, 0x6c, 0x32, 0xd3, 0x4d, - 0x6a, 0xae, 0x92, 0x85, 0xed, 0x26, 0xd4, 0x35, 0x1c, 0xbd, 0x98, 0xf9, 0xdc, 0xb1, 0x37, 0xed, - 0x56, 0xcd, 0x03, 0xbd, 0x75, 0xc8, 0x7c, 0x4e, 0x6e, 0xc1, 0x2a, 0xc3, 0x01, 0xfa, 0x67, 0x38, - 0xec, 0x51, 0xd1, 0x0b, 0xb8, 0xc2, 0xca, 0xf6, 0x1a, 0xc9, 0xee, 0x43, 0xb1, 0xcf, 0x65, 0x0e, - 0x5c, 0x50, 0x11, 0x73, 0x67, 0x51, 0xe7, 0xa0, 0x57, 0x64, 0x03, 0xe0, 0x84, 0x72, 0xd1, 0x43, - 0xc6, 0x22, 0xe6, 0x54, 0xd5, 0x59, 0x4d, 0xee, 0xec, 0xc9, 0x0d, 0xe2, 0xc1, 0x72, 0x80, 0x82, - 0x0e, 0xa9, 0xa0, 0xce, 0xd2, 0xa6, 0xdd, 0xaa, 0xef, 0x7c, 0xd1, 0x2e, 0xba, 0x0e, 0xed, 0x7c, - 0x8e, 0xed, 0x7d, 0xa3, 0xb8, 0x17, 0x0a, 0x36, 0xf1, 0x52, 0x3b, 0xcd, 0x7b, 0xb0, 0x92, 0x3b, - 0x22, 0x6b, 0x60, 0x1f, 0xe3, 0xc4, 0x60, 0x21, 0x3f, 0x25, 0x14, 0x67, 0xf4, 0x64, 0x0a, 0x85, - 0x5a, 0x7c, 0x5d, 0xf9, 0xca, 0x72, 0x9f, 0xc1, 0xfa, 0x53, 0x14, 0x79, 0x4f, 0xbb, 0x93, 0xe7, - 0x8f, 0xcb, 0xf8, 0x9b, 0x09, 0xac, 0x3b, 0x80, 0x8d, 0x02, 0x4b, 0xa6, 0x5e, 0x76, 0x61, 0x89, - 0xe9, 0x53, 0x65, 0xad, 0xbe, 0xd3, 0x9a, 0x37, 0x75, 0x2f, 0x51, 0x74, 0x5f, 0xc1, 0xc7, 0x33, - 0x9c, 0xe8, 0x3a, 0x3f, 0xf4, 0x9e, 0x27, 0x51, 0x6f, 0x00, 0x4c, 0x49, 0x36, 0xb1, 0xd7, 0x52, - 0x8e, 0x0b, 0x12, 0x38, 0x81, 0x5b, 0xe5, 0xb6, 0x4d, 0x1e, 0x8f, 0x61, 0xd9, 0x84, 0xc3, 0x1d, - 0x4b, 0x71, 0x38, 0x7f, 0x22, 0xa9, 0xa6, 0xfb, 0xa7, 0xbc, 0xe4, 0x3e, 0x17, 0xe5, 0x97, 0xfc, - 0x2e, 0x7c, 0x98, 0x2d, 0xc6, 0x88, 0xf5, 0xe8, 0x6b, 0x81, 0x4c, 0x56, 0x65, 0x45, 0x55, 0xe5, - 0xfb, 0xd3, 0xaa, 0xfc, 0x96, 0x3d, 0x94, 0x67, 0xfb, 0x9c, 0x7c, 0x0a, 0x24, 0xd5, 0xea, 0xe3, - 0xeb, 0x88, 0xa1, 0x54, 0xd0, 0x3d, 0x68, 0x2d, 0x39, 0xd9, 0x55, 0x07, 0xfb, 0x9c, 0x7c, 0x04, - 0xb5, 0x31, 0x1d, 0x61, 0x8f, 0xfb, 0xe7, 0xba, 0x17, 0x2d, 0x7a, 0xcb, 0x72, 0xe3, 0xc0, 0x3f, - 0x47, 0x89, 0xa4, 0x3a, 0x14, 0xd1, 0x31, 0x86, 0xa6, 0xd6, 0x95, 0xf8, 0x77, 0x72, 0xc3, 0xfd, - 0x19, 0x1a, 0x3a, 0x89, 0x77, 0x89, 0x0d, 0xd9, 0x86, 0xf7, 0x42, 0xfc, 0x49, 0xf4, 0x32, 0x9e, - 0x35, 0x53, 0x2b, 0x72, 0xbb, 0x9b, 0x7a, 0xcf, 0x15, 0xef, 0x33, 0x9f, 0x8b, 0xe8, 0xaa, 0xc5, - 0xfb, 0x8f, 0x05, 0x0d, 0x63, 0x60, 0xef, 0x0c, 0x43, 0x21, 0x1b, 0x7c, 0xda, 0xac, 0x25, 0x78, - 0x96, 0x02, 0xaf, 0x9e, 0xee, 0xe5, 0x5a, 0x40, 0xa5, 0xa4, 0x05, 0xd8, 0x17, 0x5b, 0x40, 0x37, - 0xd3, 0x02, 0x16, 0x14, 0x44, 0x77, 0x8b, 0x21, 0xca, 0xc6, 0xf4, 0xff, 0x34, 0x80, 0x5e, 0xf6, - 0xda, 0xe6, 0x30, 0x34, 0x94, 0x3e, 0x80, 0x2a, 0x4a, 0xf7, 0x09, 0xa1, 0xdb, 0xf3, 0x45, 0xeb, - 0x19, 0xad, 0xfc, 0x95, 0x4d, 0x1d, 0xbc, 0x9b, 0x2b, 0x3b, 0x4c, 0x1f, 0x02, 0x63, 0x78, 0x26, - 0xe5, 0xd3, 0x0c, 0x2a, 0x57, 0xca, 0x20, 0xcc, 0x36, 0x86, 0x59, 0x19, 0x18, 0xa4, 0x9e, 0x40, - 0xed, 0x48, 0x9d, 0xfa, 0x38, 0x7f, 0xf5, 0x1b, 0x7b, 0xde, 0x54, 0xd5, 0xdd, 0x82, 0x45, 0x5d, - 0x2a, 0xc5, 0xa3, 0x0a, 0xc2, 0x8d, 0xc3, 0x90, 0xe1, 0x20, 0x1a, 0x85, 0xfe, 0x39, 0x0e, 0x5f, - 0x4a, 0xcb, 0x5a, 0xe7, 0x73, 0x58, 0xd4, 0x85, 0xa7, 0x7b, 0xec, 0xcd, 0xe2, 0x00, 0x94, 0xbc, - 0xa7, 0xa5, 0x0b, 0xf0, 0xfd, 0xc5, 0x82, 0xeb, 0x26, 0xce, 0x17, 0x91, 0x78, 0x12, 0xc5, 0xe1, - 0xf0, 0x3f, 0x79, 0x49, 0xd8, 0xa9, 0x64, 0xd8, 0xc9, 0x13, 0x6f, 0x5f, 0x20, 0x7e, 0xe7, 0xef, - 0x25, 0x20, 0x07, 0xca, 0xae, 0x4a, 0xf2, 0xa9, 0x36, 0x4b, 0xbe, 0x87, 0x05, 0x39, 0xaa, 0x91, - 0x4f, 0x8a, 0x3d, 0x67, 0x86, 0xbe, 0xe6, 0xf6, 0x9b, 0xc4, 0xcc, 0x68, 0x71, 0x4d, 0x1a, 0x96, - 0x53, 0x4e, 0x99, 0xe1, 0xcc, 0x6c, 0x56, 0x66, 0x38, 0x3b, 0x2c, 0xb9, 0xd7, 0xc8, 0x0f, 0x50, - 0xd5, 0x73, 0x0c, 0xb9, 0x5d, 0xac, 0x93, 0x1b, 0x9e, 0x9a, 0xad, 0x37, 0x0b, 0xa6, 0xe6, 0x7f, - 0xb3, 0xe0, 0x83, 0x99, 0xef, 0x2f, 0x29, 0x99, 0x30, 0xca, 0x9e, 0xfe, 0xe6, 0x97, 0x6f, 0xad, - 0x97, 0x06, 0xf3, 0x87, 0x35, 0x73, 0xac, 0x48, 0xaf, 0x0c, 0xb9, 0xff, 0x56, 0xb6, 0x2f, 0x36, - 0x8b, 0xe6, 0x83, 0xab, 0xaa, 0xe7, 0x68, 0xf6, 0xb9, 0x28, 0xa5, 0x79, 0xfa, 0x3a, 0x97, 0xd2, - 0x9c, 0x79, 0xff, 0x2e, 0xf1, 0x90, 0x69, 0xa8, 0xf3, 0xf1, 0x70, 0xf9, 0x15, 0x9b, 0x8f, 0x87, - 0x19, 0x9d, 0xfb, 0x12, 0x0f, 0x97, 0x5b, 0xd7, 0x7c, 0x3c, 0x14, 0x36, 0xed, 0xf9, 0x78, 0x28, - 0xee, 0x98, 0xee, 0xb5, 0xdd, 0x63, 0x58, 0x1f, 0x44, 0x41, 0xa1, 0x99, 0xdd, 0x86, 0xb9, 0xf0, - 0x5d, 0xf9, 0x57, 0xae, 0x6b, 0xbd, 0xba, 0x37, 0xf2, 0xc5, 0x51, 0xdc, 0x6f, 0x0f, 0xa2, 0xa0, - 0x23, 0x95, 0x3a, 0x19, 0xa5, 0x8e, 0xfc, 0x6b, 0x98, 0x5d, 0x1b, 0x23, 0xfa, 0xbf, 0xe1, 0xb8, - 0xdf, 0xaf, 0xaa, 0x8f, 0xcf, 0xfe, 0x0d, 0x00, 0x00, 0xff, 0xff, 0x54, 0xbb, 0x2f, 0xff, 0x91, - 0x0e, 0x00, 0x00, + 0x3d, 0xf4, 0xd2, 0x4b, 0xcf, 0xbd, 0x17, 0xfd, 0x2d, 0x3d, 0xf5, 0x3f, 0xf4, 0xc7, 0x14, 0xfb, + 0x20, 0x45, 0xda, 0x22, 0xa3, 0x38, 0x39, 0x89, 0xbb, 0x3b, 0xcf, 0xef, 0x9b, 0x9d, 0x1d, 0xc1, + 0xca, 0x88, 0x0a, 0xfc, 0x99, 0x4e, 0xda, 0x63, 0x16, 0x89, 0x88, 0x38, 0x71, 0x1f, 0x59, 0x9b, + 0xc7, 0xfd, 0xc0, 0x17, 0xa7, 0x31, 0xc6, 0xd8, 0x36, 0xe7, 0x4d, 0x97, 0x8e, 0xfd, 0x4e, 0x9f, + 0x72, 0xec, 0x0c, 0x8e, 0x68, 0x38, 0xc2, 0x8e, 0x52, 0x30, 0x0b, 0xad, 0xdd, 0xbc, 0x93, 0xca, + 0x04, 0xc8, 0x46, 0xc8, 0x05, 0xa3, 0x02, 0x47, 0x13, 0x23, 0x9a, 0xdb, 0xd3, 0x1a, 0xee, 0x6d, + 0xa8, 0x77, 0xfd, 0x70, 0xe4, 0xe1, 0x69, 0x8c, 0x5c, 0x10, 0x07, 0x96, 0x02, 0xe4, 0x9c, 0x8e, + 0xd0, 0xb1, 0x36, 0xad, 0x56, 0xcd, 0x4b, 0x96, 0xee, 0xef, 0x16, 0x34, 0xb4, 0x24, 0x1f, 0x47, + 0x21, 0xc7, 0x62, 0x51, 0xb2, 0x05, 0x0d, 0x8e, 0xec, 0xcc, 0x1f, 0x60, 0x2f, 0xa4, 0x01, 0x3a, + 0x15, 0x75, 0x5c, 0x37, 0x7b, 0x2f, 0x68, 0x80, 0x64, 0x1d, 0x6a, 0xc2, 0x0f, 0x90, 0x0b, 0x1a, + 0x8c, 0x1d, 0x7b, 0xd3, 0x6a, 0xd9, 0xde, 0x74, 0x83, 0x34, 0x61, 0xf9, 0x28, 0xe2, 0x42, 0x29, + 0x2f, 0x28, 0xe5, 0x74, 0xed, 0xfe, 0x69, 0x41, 0xfd, 0x3b, 0x1a, 0x0e, 0x93, 0x88, 0xaf, 0xc3, + 0xa2, 0xc2, 0xc9, 0x04, 0xa1, 0x17, 0xe4, 0x0e, 0x54, 0x35, 0x30, 0xca, 0x79, 0x7d, 0xc7, 0x69, + 0x2b, 0x5c, 0x25, 0x34, 0x6d, 0x83, 0xd8, 0x23, 0xf5, 0xe3, 0x19, 0x39, 0x72, 0x1f, 0x96, 0x13, + 0x68, 0x94, 0xcf, 0xd5, 0x9d, 0xad, 0x8c, 0x4e, 0x1e, 0xba, 0x03, 0xf3, 0xe1, 0xa5, 0x2a, 0xae, + 0x0b, 0x0d, 0x1d, 0x95, 0x41, 0x87, 0xc0, 0x02, 0x3f, 0xf5, 0x87, 0x26, 0x2a, 0xf5, 0xed, 0xbe, + 0x84, 0x95, 0x47, 0x34, 0x1c, 0xe0, 0x49, 0x12, 0xfb, 0x0c, 0x21, 0x72, 0x03, 0xaa, 0x0c, 0x29, + 0x8f, 0x42, 0x03, 0x9b, 0x59, 0x4d, 0xf3, 0xb4, 0x33, 0x79, 0xba, 0x6b, 0xb0, 0x9a, 0x98, 0xd4, + 0x8e, 0xdd, 0x7f, 0x2b, 0xb0, 0x6a, 0xec, 0x1f, 0xc4, 0x41, 0x40, 0xd9, 0x64, 0xa6, 0x9b, 0xd4, + 0x5c, 0x25, 0x0b, 0xdb, 0x4d, 0xa8, 0x6b, 0x38, 0x7a, 0x31, 0xf3, 0xb9, 0x63, 0x6f, 0xda, 0xad, + 0x9a, 0x07, 0x7a, 0xeb, 0x90, 0xf9, 0x9c, 0xdc, 0x82, 0x55, 0x86, 0x03, 0xf4, 0xcf, 0x70, 0xd8, + 0xa3, 0xa2, 0x17, 0x70, 0x85, 0x95, 0xed, 0x35, 0x92, 0xdd, 0x87, 0x62, 0x9f, 0xcb, 0x1c, 0xb8, + 0xa0, 0x22, 0xe6, 0xce, 0xa2, 0xce, 0x41, 0xaf, 0xc8, 0x06, 0xc0, 0x09, 0xe5, 0xa2, 0x87, 0x8c, + 0x45, 0xcc, 0xa9, 0xaa, 0xb3, 0x9a, 0xdc, 0xd9, 0x93, 0x1b, 0xc4, 0x83, 0xe5, 0x00, 0x05, 0x1d, + 0x52, 0x41, 0x9d, 0xa5, 0x4d, 0xbb, 0x55, 0xdf, 0xf9, 0xaa, 0x5d, 0x74, 0x1d, 0xda, 0xf9, 0x1c, + 0xdb, 0xfb, 0x46, 0x71, 0x2f, 0x14, 0x6c, 0xe2, 0xa5, 0x76, 0x9a, 0xf7, 0x60, 0x25, 0x77, 0x44, + 0xd6, 0xc0, 0x3e, 0xc6, 0x89, 0xc1, 0x42, 0x7e, 0x4a, 0x28, 0xce, 0xe8, 0xc9, 0x14, 0x0a, 0xb5, + 0xf8, 0xb6, 0xf2, 0x8d, 0xe5, 0x3e, 0x83, 0xf5, 0xa7, 0x28, 0xf2, 0x9e, 0x76, 0x27, 0xcf, 0x1f, + 0x97, 0xf1, 0x37, 0x13, 0x58, 0x77, 0x00, 0x1b, 0x05, 0x96, 0x4c, 0xbd, 0xec, 0xc2, 0x12, 0xd3, + 0xa7, 0xca, 0x5a, 0x7d, 0xa7, 0x35, 0x6f, 0xea, 0x5e, 0xa2, 0xe8, 0xbe, 0x82, 0x4f, 0x67, 0x38, + 0xd1, 0x75, 0x7e, 0xe8, 0x3d, 0x4f, 0xa2, 0xde, 0x00, 0x98, 0x92, 0x6c, 0x62, 0xaf, 0xa5, 0x1c, + 0x17, 0x24, 0x70, 0x02, 0xb7, 0xca, 0x6d, 0x9b, 0x3c, 0x1e, 0xc3, 0xb2, 0x09, 0x87, 0x3b, 0x96, + 0xe2, 0x70, 0xfe, 0x44, 0x52, 0x4d, 0xf7, 0x1f, 0x79, 0xc9, 0x7d, 0x2e, 0xca, 0x2f, 0xf9, 0x5d, + 0xf8, 0x38, 0x5b, 0x8c, 0x11, 0xeb, 0xd1, 0xd7, 0x02, 0x99, 0xac, 0xca, 0x8a, 0xaa, 0xca, 0x0f, + 0xa7, 0x55, 0xf9, 0x3d, 0x7b, 0x28, 0xcf, 0xf6, 0x39, 0xf9, 0x1c, 0x48, 0xaa, 0xd5, 0xc7, 0xd7, + 0x11, 0x43, 0xa9, 0xa0, 0x7b, 0xd0, 0x5a, 0x72, 0xb2, 0xab, 0x0e, 0xf6, 0x39, 0xf9, 0x04, 0x6a, + 0x63, 0x3a, 0xc2, 0x1e, 0xf7, 0xcf, 0x75, 0x2f, 0x5a, 0xf4, 0x96, 0xe5, 0xc6, 0x81, 0x7f, 0x8e, + 0x12, 0x49, 0x75, 0x28, 0xa2, 0x63, 0x0c, 0x4d, 0xad, 0x2b, 0xf1, 0x1f, 0xe4, 0x86, 0xfb, 0x2b, + 0x34, 0x74, 0x12, 0xef, 0x13, 0x1b, 0xb2, 0x0d, 0x1f, 0x84, 0xf8, 0x8b, 0xe8, 0x65, 0x3c, 0x6b, + 0xa6, 0x56, 0xe4, 0x76, 0x37, 0xf5, 0x9e, 0x2b, 0xde, 0x67, 0x3e, 0x17, 0xd1, 0x55, 0x8b, 0xf7, + 0xef, 0x0a, 0x34, 0x8c, 0x81, 0xbd, 0x33, 0x0c, 0x85, 0x6c, 0xf0, 0x69, 0xb3, 0x96, 0xe0, 0x59, + 0x0a, 0xbc, 0x7a, 0xba, 0x97, 0x6b, 0x01, 0x95, 0x92, 0x16, 0x60, 0x5f, 0x6c, 0x01, 0xdd, 0x4c, + 0x0b, 0x58, 0x50, 0x10, 0xdd, 0x2d, 0x86, 0x28, 0x1b, 0x53, 0x51, 0x03, 0x90, 0x69, 0x8a, 0xc9, + 0x18, 0x0d, 0x3b, 0xea, 0x5b, 0xa6, 0x89, 0x52, 0xc9, 0xb4, 0x20, 0xbd, 0x78, 0xb7, 0x56, 0xd1, + 0xcb, 0x5e, 0xf0, 0x1c, 0xda, 0x86, 0xfc, 0x07, 0x50, 0x55, 0x6e, 0x12, 0xea, 0xb7, 0xe7, 0xcb, + 0xcb, 0x33, 0x5a, 0xf9, 0xcb, 0x9d, 0x3a, 0x78, 0x3f, 0x97, 0x7b, 0x98, 0x3e, 0x19, 0xc6, 0xf0, + 0xcc, 0xe2, 0x98, 0x66, 0x50, 0xb9, 0x52, 0x06, 0x61, 0xb6, 0x85, 0xcc, 0xca, 0xc0, 0x20, 0xf5, + 0x04, 0x6a, 0x47, 0xea, 0xd4, 0xc7, 0xf9, 0xef, 0x89, 0xb1, 0xe7, 0x4d, 0x55, 0xdd, 0x2d, 0x58, + 0xd4, 0x45, 0x55, 0x3c, 0xd4, 0x20, 0xdc, 0x38, 0x0c, 0x19, 0x0e, 0xa2, 0x51, 0xe8, 0x9f, 0xe3, + 0xf0, 0xa5, 0xb4, 0xac, 0x75, 0xbe, 0x84, 0x45, 0x5d, 0xa2, 0xba, 0x1b, 0xdf, 0x2c, 0x0e, 0x40, + 0xc9, 0x7b, 0x5a, 0xba, 0x00, 0xdf, 0xdf, 0x2c, 0xb8, 0x6e, 0xe2, 0x7c, 0x11, 0x89, 0x27, 0x51, + 0x1c, 0x0e, 0xdf, 0xc9, 0x4b, 0xc2, 0x4e, 0x25, 0xc3, 0x4e, 0x9e, 0x78, 0xfb, 0x02, 0xf1, 0x3b, + 0xff, 0x2d, 0x01, 0x39, 0x50, 0x76, 0x55, 0x92, 0x4f, 0xb5, 0x59, 0xf2, 0x23, 0x2c, 0xc8, 0xa1, + 0x8e, 0x7c, 0x56, 0xec, 0x39, 0x33, 0x1e, 0x36, 0xb7, 0xdf, 0x24, 0x66, 0x86, 0x90, 0x6b, 0xd2, + 0xb0, 0x9c, 0x87, 0xca, 0x0c, 0x67, 0xa6, 0xb8, 0x32, 0xc3, 0xd9, 0xb1, 0xca, 0xbd, 0x46, 0x7e, + 0x82, 0xaa, 0x9e, 0x78, 0xc8, 0xed, 0x62, 0x9d, 0xdc, 0x98, 0xd5, 0x6c, 0xbd, 0x59, 0x30, 0x35, + 0xff, 0x87, 0x05, 0x1f, 0xcd, 0x7c, 0xa9, 0x49, 0xc9, 0x2c, 0x52, 0x36, 0x24, 0x34, 0xbf, 0x7e, + 0x6b, 0xbd, 0x34, 0x98, 0xbf, 0xac, 0x99, 0x03, 0x48, 0x7a, 0x65, 0xc8, 0xfd, 0xb7, 0xb2, 0x7d, + 0xb1, 0x59, 0x34, 0x1f, 0x5c, 0x55, 0x3d, 0x47, 0xb3, 0xcf, 0x45, 0x29, 0xcd, 0xd3, 0x77, 0xbc, + 0x94, 0xe6, 0xcc, 0x4b, 0x79, 0x89, 0x87, 0x4c, 0x43, 0x9d, 0x8f, 0x87, 0xcb, 0xef, 0xdd, 0x7c, + 0x3c, 0xcc, 0xe8, 0xdc, 0x97, 0x78, 0xb8, 0xdc, 0xba, 0xe6, 0xe3, 0xa1, 0xb0, 0x69, 0xcf, 0xc7, + 0x43, 0x71, 0xc7, 0x74, 0xaf, 0xed, 0x1e, 0xc3, 0xfa, 0x20, 0x0a, 0x0a, 0xcd, 0xec, 0x36, 0xcc, + 0x85, 0xef, 0xca, 0x3f, 0x7d, 0x5d, 0xeb, 0xd5, 0xbd, 0x91, 0x2f, 0x8e, 0xe2, 0x7e, 0x7b, 0x10, + 0x05, 0x1d, 0xa9, 0xd4, 0xc9, 0x28, 0x75, 0xe4, 0x9f, 0xc8, 0xec, 0xda, 0x18, 0xd1, 0xff, 0x22, + 0xc7, 0xfd, 0x7e, 0x55, 0x7d, 0x7c, 0xf1, 0x7f, 0x00, 0x00, 0x00, 0xff, 0xff, 0x9c, 0x51, 0x24, + 0x34, 0xbb, 0x0e, 0x00, 0x00, }, // api/base/change/proto/change.proto []byte{ diff --git a/service/submitqueue/gateway/server/mapper/request_history.go b/service/submitqueue/gateway/server/mapper/request_history.go index 89ff1423..4a674a2e 100644 --- a/service/submitqueue/gateway/server/mapper/request_history.go +++ b/service/submitqueue/gateway/server/mapper/request_history.go @@ -30,12 +30,18 @@ func ProtoToGetRequestHistoryByChangeURIRequest(req *pb.GetRequestHistoryByChang } // HistoryEventsToProto maps retained request-log events to wire history events. +// +// Status and Event are mutually exclusive, mirroring the entry itself: a client +// reading the timeline can render a position and an occurrence differently +// without having to know which values belong to which vocabulary. func HistoryEventsToProto(logs []entity.RequestLog) []*pb.HistoryEvent { events := make([]*pb.HistoryEvent, len(logs)) for i, log := range logs { events[i] = &pb.HistoryEvent{ TimestampMs: log.TimestampMs, + Type: string(log.Type), Status: string(log.Status), + Event: string(log.Event), LastError: log.LastError, Metadata: cloneStringMap(log.Metadata), } diff --git a/submitqueue/core/request/log.go b/submitqueue/core/request/log.go index eef6615e..b47342b7 100644 --- a/submitqueue/core/request/log.go +++ b/submitqueue/core/request/log.go @@ -27,18 +27,30 @@ import ( // PublishLog publishes a single request log entry to the log topic for async persistence. // The partitionKey ensures ordering of log entries for the same request; typically set to the request ID. // -// The message ID is scoped to (requestID, status) so that the queue's +// The message ID is scoped to (requestID, status, occurrence) so that the queue's // (topic, partition_key, id) unique index dedupes retries of the same logical // log event (same delivery re-processed) without rejecting distinct statuses // for the same request (e.g. "started" emitted by the start controller and // "cancelled" emitted later by the cancel controller). -func PublishLog(ctx context.Context, registry consumer.TopicRegistry, logEntry entity.RequestLog, partitionKey string) error { +// +// occurrence is a stable discriminator naming this occurrence of the status. +// Empty means the status happens at most once per request, and the id collapses +// to (requestID, status) — every repeat is then dropped at publish time, which +// is what a terminal status wants. A status that legitimately recurs (a request +// builds again each time speculation re-plans) must pass something that is +// stable across redeliveries of one occurrence but differs between occurrences: +// the build ID, or the batch and path the entry is about. Deriving it from the +// wall clock or a random value would defeat the dedupe entirely. +func PublishLog(ctx context.Context, registry consumer.TopicRegistry, logEntry entity.RequestLog, partitionKey string, occurrence string) error { payload, err := logEntry.ToBytes() if err != nil { return fmt.Errorf("failed to serialize request log: %w", err) } - msgID := fmt.Sprintf("%s/%s", logEntry.RequestID, logEntry.Status) + msgID := fmt.Sprintf("%s/%s", logEntry.RequestID, logEntry.Value()) + if occurrence != "" { + msgID = fmt.Sprintf("%s/%s", msgID, occurrence) + } msg := entityqueue.NewMessage(msgID, payload, partitionKey, nil) q, ok := registry.Queue(topickey.TopicKeyLog) @@ -58,15 +70,39 @@ func PublishLog(ctx context.Context, registry consumer.TopicRegistry, logEntry e return nil } -// PublishBatchLogs publishes a request log entry for each request ID in the batch to the log topic. +// PublishBatchLogs publishes a status log entry for each request ID in the batch to the log topic. // Each entry uses the request ID as the partition key to ensure per-request ordering. // queue scopes every entry: a request ID is only unique within its own queue. -func PublishBatchLogs(ctx context.Context, registry consumer.TopicRegistry, queue string, requestIDs []string, status entity.RequestStatus, metadata map[string]string) error { +// occurrence names this occurrence of the status and is shared by every entry in +// the fan-out, so the whole batch dedupes together on a redelivery; see PublishLog. +// +// Entries carry no request version: a status published for a whole batch reports +// something that happened to the batch, not a transition of any one request's +// state machine. +func PublishBatchLogs(ctx context.Context, registry consumer.TopicRegistry, queue string, requestIDs []string, status entity.RequestStatus, occurrence string, metadata map[string]string) error { for _, requestID := range requestIDs { - logEntry := entity.NewRequestLog(queue, requestID, status, 0, "", metadata) - if err := PublishLog(ctx, registry, logEntry, requestID); err != nil { + logEntry := entity.NewRequestStatusLog(queue, requestID, status, 0, "", metadata) + if err := PublishLog(ctx, registry, logEntry, requestID, occurrence); err != nil { return fmt.Errorf("failed to publish request log for request %s: %w", requestID, err) } } return nil } + +// PublishBatchEvents publishes an event log entry for each request ID in the batch +// to the log topic. It is the event counterpart of PublishBatchLogs and shares its +// partitioning and occurrence semantics. +// +// Separate rather than a flag on PublishBatchLogs because the two carry different +// vocabularies: a caller reporting build progress cannot reach for a status, and a +// caller reporting a status cannot accidentally publish something the summary will +// refuse to project. +func PublishBatchEvents(ctx context.Context, registry consumer.TopicRegistry, queue string, requestIDs []string, event entity.RequestEvent, occurrence string, metadata map[string]string) error { + for _, requestID := range requestIDs { + logEntry := entity.NewRequestEventLog(queue, requestID, event, metadata) + if err := PublishLog(ctx, registry, logEntry, requestID, occurrence); err != nil { + return fmt.Errorf("failed to publish request event for request %s: %w", requestID, err) + } + } + return nil +} diff --git a/submitqueue/core/request/log_test.go b/submitqueue/core/request/log_test.go index 60c911ff..ea331479 100644 --- a/submitqueue/core/request/log_test.go +++ b/submitqueue/core/request/log_test.go @@ -50,8 +50,8 @@ func TestPublishLog_Success(t *testing.T) { ctrl := gomock.NewController(t) registry := newTestRegistry(t, ctrl, nil) - logEntry := entity.NewRequestLog("req", "req/1", entity.RequestStatusStarted, 1, "", nil) - err := PublishLog(context.Background(), registry, logEntry, "req/1") + logEntry := entity.NewRequestStatusLog("req", "req/1", entity.RequestStatusStarted, 1, "", nil) + err := PublishLog(context.Background(), registry, logEntry, "req/1", "") require.NoError(t, err) } @@ -59,8 +59,8 @@ func TestPublishLog_PublishFailure(t *testing.T) { ctrl := gomock.NewController(t) registry := newTestRegistry(t, ctrl, fmt.Errorf("connection refused")) - logEntry := entity.NewRequestLog("req", "req/1", entity.RequestStatusStarted, 1, "", nil) - err := PublishLog(context.Background(), registry, logEntry, "req/1") + logEntry := entity.NewRequestStatusLog("req", "req/1", entity.RequestStatusStarted, 1, "", nil) + err := PublishLog(context.Background(), registry, logEntry, "req/1", "") require.Error(t, err) } @@ -70,6 +70,7 @@ func TestPublishBatchLogs_Success(t *testing.T) { err := PublishBatchLogs(context.Background(), registry, "req", []string{"req/1", "req/2", "req/3"}, entity.RequestStatusBatched, + "", map[string]string{"batch_id": "b/1"}, ) require.NoError(t, err) @@ -100,6 +101,7 @@ func TestPublishBatchLogs_PartialFailure(t *testing.T) { err = PublishBatchLogs(context.Background(), registry, "req", []string{"req/1", "req/2", "req/3"}, entity.RequestStatusBatched, + "", map[string]string{"batch_id": "b/1"}, ) require.Error(t, err) @@ -109,7 +111,7 @@ func TestPublishBatchLogs_Empty(t *testing.T) { ctrl := gomock.NewController(t) registry := newTestRegistry(t, ctrl, nil) - err := PublishBatchLogs(context.Background(), registry, "req", nil, entity.RequestStatusBatched, nil) + err := PublishBatchLogs(context.Background(), registry, "req", nil, entity.RequestStatusBatched, "", nil) require.NoError(t, err) } @@ -147,11 +149,11 @@ func TestPublishLog_MessageIDScopedByStatus(t *testing.T) { entity.RequestStatusCancelled, } { require.NoError(t, PublishLog(context.Background(), registry, - entity.NewRequestLog("req", "req/1", st, 0, "", nil), "req/1")) + entity.NewRequestStatusLog("req", "req/1", st, 0, "", nil), "req/1", "")) } // Re-emit "started" to simulate a retry of the same delivery — must reuse the same ID. require.NoError(t, PublishLog(context.Background(), registry, - entity.NewRequestLog("req", "req/1", entity.RequestStatusStarted, 0, "", nil), "req/1")) + entity.NewRequestStatusLog("req", "req/1", entity.RequestStatusStarted, 0, "", nil), "req/1", "")) require.Equal(t, []string{ "req/1/started", @@ -160,3 +162,101 @@ func TestPublishLog_MessageIDScopedByStatus(t *testing.T) { "req/1/started", }, ids) } + +// TestPublishLog_MessageIDScopedByOccurrence locks in the recurring half of the +// id scheme. An entry that is recorded more than once per request (a request +// builds again every time speculation re-plans) is told apart by its occurrence, +// so each occurrence reaches the log while a redelivery of any one of them still +// dedupes. Without this, the second occurrence is silently swallowed by the +// queue's (topic, partition_key, id) index and a rebuilding request reads +// identically to one that built cleanly first try. +// +// The id is built from what the entry recorded, whichever kind that is, so the +// table carries whole entries rather than statuses. +func TestPublishLog_MessageIDScopedByOccurrence(t *testing.T) { + tests := []struct { + name string + entry entity.RequestLog + occurrence string + wantID string + }{ + { + name: "empty occurrence collapses to request and status", + entry: entity.NewRequestStatusLog("req", "req/1", entity.RequestStatusStarted, 0, "", nil), + occurrence: "", + wantID: "req/1/started", + }, + { + name: "build id separates one build from the next", + entry: entity.NewRequestEventLog("req", "req/1", entity.RequestEventBuilding, nil), + occurrence: "build/7", + wantID: "req/1/building/build/7", + }, + { + name: "batch and path separate one passed path from the next", + entry: entity.NewRequestStatusLog("req", "req/1", entity.RequestStatusSpeculated, 0, "", nil), + occurrence: "batch/2/path/a", + wantID: "req/1/speculated/batch/2/path/a", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + ctrl := gomock.NewController(t) + + var ids []string + mockPub := queuemock.NewMockPublisher(ctrl) + mockPub.EXPECT().Publish(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn( + func(_ context.Context, _ string, msg entityqueue.Message) error { + ids = append(ids, msg.ID) + return nil + }, + ).AnyTimes() + mockQ := queuemock.NewMockQueue(ctrl) + mockQ.EXPECT().Publisher().Return(mockPub).AnyTimes() + registry, err := consumer.NewTopicRegistry( + []consumer.TopicConfig{{Key: topickey.TopicKeyLog, Name: "log", Queue: mockQ}}, + ) + require.NoError(t, err) + + logEntry := tt.entry + // Twice: a redelivery of one occurrence must reuse its ID so the + // queue dedupes it. + require.NoError(t, PublishLog(context.Background(), registry, logEntry, "req/1", tt.occurrence)) + require.NoError(t, PublishLog(context.Background(), registry, logEntry, "req/1", tt.occurrence)) + + require.Equal(t, []string{tt.wantID, tt.wantID}, ids) + }) + } +} + +// TestPublishBatchLogs_SharesOccurrenceAcrossFanout checks that every entry of +// one fan-out carries the same occurrence, so the whole batch dedupes together +// when the publishing delivery is retried, while the per-request partition keys +// keep the entries distinct. +func TestPublishBatchLogs_SharesOccurrenceAcrossFanout(t *testing.T) { + ctrl := gomock.NewController(t) + + var ids []string + mockPub := queuemock.NewMockPublisher(ctrl) + mockPub.EXPECT().Publish(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn( + func(_ context.Context, _ string, msg entityqueue.Message) error { + ids = append(ids, msg.ID) + return nil + }, + ).AnyTimes() + mockQ := queuemock.NewMockQueue(ctrl) + mockQ.EXPECT().Publisher().Return(mockPub).AnyTimes() + registry, err := consumer.NewTopicRegistry( + []consumer.TopicConfig{{Key: topickey.TopicKeyLog, Name: "log", Queue: mockQ}}, + ) + require.NoError(t, err) + + require.NoError(t, PublishBatchLogs(context.Background(), registry, "req", + []string{"req/1", "req/2"}, entity.RequestStatusSpeculating, "batch/9", nil)) + + require.Equal(t, []string{ + "req/1/speculating/batch/9", + "req/2/speculating/batch/9", + }, ids) +} diff --git a/submitqueue/core/request/materializer.go b/submitqueue/core/request/materializer.go index ec1318ed..d470ed41 100644 --- a/submitqueue/core/request/materializer.go +++ b/submitqueue/core/request/materializer.go @@ -143,7 +143,15 @@ func (m *Materializer) createURIMappings(ctx context.Context, stores storage.Sto // logWins keeps accepting internal, treats accepted as the lowest public state, // then applies versioned-terminal precedence and timestamp ordering. +// +// Only status entries are candidates. An event describes work happening +// underneath the request's position — a build starting or finishing on one of +// several concurrent speculation paths — so it has no status to project, and its +// Status field is unset. func logWins(log entity.RequestLog, summary entity.RequestSummary) bool { + if log.Type != entity.RequestLogTypeStatus { + return false + } if summary.Status == entity.RequestStatusAccepting { return log.Status != entity.RequestStatusAccepting } diff --git a/submitqueue/core/request/materializer_test.go b/submitqueue/core/request/materializer_test.go index 787a3963..870db1fb 100644 --- a/submitqueue/core/request/materializer_test.go +++ b/submitqueue/core/request/materializer_test.go @@ -29,7 +29,7 @@ import ( func TestMaterializer_PersistLog(t *testing.T) { base := testRequestSummary() - log := entity.RequestLog{RequestID: "q/1", TimestampMs: 20, Status: entity.RequestStatusLanded, RequestVersion: 2, Metadata: map[string]string{}} + log := entity.RequestLog{RequestID: "q/1", TimestampMs: 20, Type: entity.RequestLogTypeStatus, Status: entity.RequestStatusLanded, RequestVersion: 2, Metadata: map[string]string{}} t.Run("winning log updates both projections", func(t *testing.T) { ctrl := gomock.NewController(t) m, summaryStore, queueStore, _, logStore := materializerStores(ctrl) @@ -51,11 +51,11 @@ func TestMaterializer_PersistLog(t *testing.T) { current := base current.Status = entity.RequestStatusLanded current.RequestVersion = 0 - incoming := entity.RequestLog{RequestID: "q/1", TimestampMs: 20, Status: entity.RequestStatusProcessing, RequestVersion: 0, Metadata: map[string]string{}} + incoming := entity.RequestLog{RequestID: "q/1", TimestampMs: 20, Type: entity.RequestLogTypeStatus, Status: entity.RequestStatusSpeculating, RequestVersion: 0, Metadata: map[string]string{}} logStore.EXPECT().Insert(gomock.Any(), incoming).Return(nil) summaryStore.EXPECT().Get(gomock.Any(), "q/1").Return(current, nil) summaryStore.EXPECT().Update(gomock.Any(), gomock.Any(), int32(1), int32(2)).DoAndReturn(func(_ context.Context, updated entity.RequestSummary, _, _ int32) error { - assert.Equal(t, entity.RequestStatusProcessing, updated.Status) + assert.Equal(t, entity.RequestStatusSpeculating, updated.Status) return nil }) queueStore.EXPECT().Get(gomock.Any(), int64(10), "q/1").Return(queueSummaryFromSummary(current), nil) @@ -63,6 +63,33 @@ func TestMaterializer_PersistLog(t *testing.T) { require.NoError(t, m.PersistLog(context.Background(), incoming)) }) + t.Run("build event reaches history without moving the summary", func(t *testing.T) { + for _, buildEvent := range []entity.RequestEvent{entity.RequestEventBuilding, entity.RequestEventBuilt} { + t.Run(string(buildEvent), func(t *testing.T) { + ctrl := gomock.NewController(t) + m, summaryStore, queueStore, _, logStore := materializerStores(ctrl) + current := base + current.Status = entity.RequestStatusSpeculating + current.StatusTimestampMs = 10 + // Newer than the summary, so only the entry's type keeps it out. + event := entity.RequestLog{ + RequestID: "q/1", TimestampMs: 20, + Type: entity.RequestLogTypeEvent, Event: buildEvent, + Metadata: map[string]string{"build_id": "b/7"}, + } + + // Inserted: the entry is the request's history. + logStore.EXPECT().Insert(gomock.Any(), event).Return(nil) + summaryStore.EXPECT().Get(gomock.Any(), "q/1").Return(current, nil) + // No summaryStore.Update expectation — gomock fails the test if + // the event moves the request's current status. + queueStore.EXPECT().Get(gomock.Any(), int64(10), "q/1").Return(queueSummaryFromSummary(current), nil) + + require.NoError(t, m.PersistLog(context.Background(), event)) + }) + } + }) + t.Run("CAS conflict reloads and repairs winner", func(t *testing.T) { ctrl := gomock.NewController(t) m, summaryStore, queueStore, _, logStore := materializerStores(ctrl) @@ -189,17 +216,45 @@ func TestLogWins(t *testing.T) { }, { name: "versioned terminal beats newer unversioned status", - current: entity.RequestSummary{Status: entity.RequestStatusProcessing, StatusTimestampMs: 200}, + current: entity.RequestSummary{Status: entity.RequestStatusSpeculating, StatusTimestampMs: 200}, incoming: entity.RequestLog{ Status: entity.RequestStatusLanded, RequestVersion: 1, TimestampMs: 100, }, want: true, }, + { + // A head funds several paths and builds them at once, so one build + // starting says nothing about where the request is: it is still + // speculating. Letting this win would also freeze the summary, + // since nothing publishes again until the next position. + name: "a build event never becomes the current status", + current: entity.RequestSummary{Status: entity.RequestStatusSpeculating, StatusTimestampMs: 100}, + incoming: entity.RequestLog{ + Type: entity.RequestLogTypeEvent, Event: entity.RequestEventBuilding, TimestampMs: 200, + }, + }, + { + name: "a completed build event does not report the request finished", + current: entity.RequestSummary{Status: entity.RequestStatusSpeculating, StatusTimestampMs: 100}, + incoming: entity.RequestLog{ + Type: entity.RequestLogTypeEvent, Event: entity.RequestEventBuilt, TimestampMs: 200, + }, + }, + { + // The position after the events still moves normally, so the + // exclusion is of the events themselves, not of that whole window. + name: "the position after the build still wins", + current: entity.RequestSummary{Status: entity.RequestStatusSpeculating, StatusTimestampMs: 100}, + incoming: entity.RequestLog{ + Status: entity.RequestStatusSpeculated, TimestampMs: 200, + }, + want: true, + }, { name: "nonterminal cannot replace versioned terminal", current: entity.RequestSummary{Status: entity.RequestStatusLanded, RequestVersion: 1, StatusTimestampMs: 100}, incoming: entity.RequestLog{ - Status: entity.RequestStatusProcessing, TimestampMs: 200, + Status: entity.RequestStatusSpeculating, TimestampMs: 200, }, }, { @@ -244,7 +299,15 @@ func TestLogWins(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - assert.Equal(t, tt.want, logWins(tt.incoming, tt.current)) + // Cases above leave Type unset when the entry is a status, so the + // table reads as being about the ordering rules under test. This + // mirrors production exactly: logWins only ever sees entries that + // came through RequestLogFromBytes, which applies the same default. + incoming := tt.incoming + if incoming.Type == "" { + incoming.Type = entity.RequestLogTypeStatus + } + assert.Equal(t, tt.want, logWins(incoming, tt.current)) }) } } diff --git a/submitqueue/core/request/request.go b/submitqueue/core/request/request.go index e6bde5ee..318f2f03 100644 --- a/submitqueue/core/request/request.go +++ b/submitqueue/core/request/request.go @@ -62,6 +62,12 @@ func GetCurrentStateFromRequestLog(ctx context.Context, store storage.RequestLog // iterate over all log records, storage contract guarantees that the records are ordered by timestamp ascending. log := &logs[i] + // Events describe work happening underneath the request's position and + // carry no status, so they are not candidates for the current one. + if log.Type != entity.RequestLogTypeStatus { + continue + } + // Track the record with the largest timestamp as fallback. if bestLatest == nil || log.TimestampMs > bestLatest.TimestampMs { bestLatest = log @@ -82,6 +88,13 @@ func GetCurrentStateFromRequestLog(ctx context.Context, store storage.RequestLog winner = bestTerminal } + // No status entry to report: either the request has no log at all, or + // everything recorded so far is an event, which carries no status. Both are + // the documented not-found case, and without this the winner is nil. + if winner == nil { + return CurrentState{}, fmt.Errorf("no request status recorded for request_id=%s: %w", requestID, storage.ErrNotFound) + } + return CurrentState{ Status: winner.Status, LastError: winner.LastError, diff --git a/submitqueue/core/request/request_test.go b/submitqueue/core/request/request_test.go index 75a955f9..fe081ce4 100644 --- a/submitqueue/core/request/request_test.go +++ b/submitqueue/core/request/request_test.go @@ -37,58 +37,58 @@ func TestGetCurrentStateFromRequestLog(t *testing.T) { { name: "single record", logs: []entity.RequestLog{ - {RequestID: "q/1", TimestampMs: 1000, Status: entity.RequestStatusStarted, RequestVersion: 1, LastError: "", Metadata: map[string]string{}}, + {RequestID: "q/1", TimestampMs: 1000, Type: entity.RequestLogTypeStatus, Status: entity.RequestStatusStarted, RequestVersion: 1, LastError: "", Metadata: map[string]string{}}, }, expected: CurrentState{Status: entity.RequestStatusStarted, LastError: "", Metadata: map[string]string{}}, }, { name: "terminal status wins over later non-terminal", logs: []entity.RequestLog{ - {RequestID: "q/1", TimestampMs: 1000, Status: entity.RequestStatusStarted, RequestVersion: 1, LastError: "", Metadata: map[string]string{}}, - {RequestID: "q/1", TimestampMs: 2000, Status: entity.RequestStatusLanded, RequestVersion: 3, LastError: "", Metadata: map[string]string{"batch": "b1"}}, - {RequestID: "q/1", TimestampMs: 3000, Status: entity.RequestStatusProcessing, RequestVersion: 0, LastError: "", Metadata: map[string]string{}}, + {RequestID: "q/1", TimestampMs: 1000, Type: entity.RequestLogTypeStatus, Status: entity.RequestStatusStarted, RequestVersion: 1, LastError: "", Metadata: map[string]string{}}, + {RequestID: "q/1", TimestampMs: 2000, Type: entity.RequestLogTypeStatus, Status: entity.RequestStatusLanded, RequestVersion: 3, LastError: "", Metadata: map[string]string{"batch": "b1"}}, + {RequestID: "q/1", TimestampMs: 3000, Type: entity.RequestLogTypeStatus, Status: entity.RequestStatusSpeculating, RequestVersion: 0, LastError: "", Metadata: map[string]string{}}, }, expected: CurrentState{Status: entity.RequestStatusLanded, LastError: "", Metadata: map[string]string{"batch": "b1"}}, }, { name: "terminal error status with last error", logs: []entity.RequestLog{ - {RequestID: "q/1", TimestampMs: 1000, Status: entity.RequestStatusStarted, RequestVersion: 1, LastError: "", Metadata: map[string]string{}}, - {RequestID: "q/1", TimestampMs: 2000, Status: entity.RequestStatusError, RequestVersion: 4, LastError: "merge conflict", Metadata: map[string]string{"step": "merge"}}, + {RequestID: "q/1", TimestampMs: 1000, Type: entity.RequestLogTypeStatus, Status: entity.RequestStatusStarted, RequestVersion: 1, LastError: "", Metadata: map[string]string{}}, + {RequestID: "q/1", TimestampMs: 2000, Type: entity.RequestLogTypeStatus, Status: entity.RequestStatusError, RequestVersion: 4, LastError: "merge conflict", Metadata: map[string]string{"step": "merge"}}, }, expected: CurrentState{Status: entity.RequestStatusError, LastError: "merge conflict", Metadata: map[string]string{"step": "merge"}}, }, { name: "multiple terminal records picks highest version", logs: []entity.RequestLog{ - {RequestID: "q/1", TimestampMs: 1000, Status: entity.RequestStatusError, RequestVersion: 2, LastError: "timeout", Metadata: map[string]string{}}, - {RequestID: "q/1", TimestampMs: 2000, Status: entity.RequestStatusLanded, RequestVersion: 5, LastError: "", Metadata: map[string]string{"final": "true"}}, - {RequestID: "q/1", TimestampMs: 3000, Status: entity.RequestStatusError, RequestVersion: 3, LastError: "conflict", Metadata: map[string]string{}}, + {RequestID: "q/1", TimestampMs: 1000, Type: entity.RequestLogTypeStatus, Status: entity.RequestStatusError, RequestVersion: 2, LastError: "timeout", Metadata: map[string]string{}}, + {RequestID: "q/1", TimestampMs: 2000, Type: entity.RequestLogTypeStatus, Status: entity.RequestStatusLanded, RequestVersion: 5, LastError: "", Metadata: map[string]string{"final": "true"}}, + {RequestID: "q/1", TimestampMs: 3000, Type: entity.RequestLogTypeStatus, Status: entity.RequestStatusError, RequestVersion: 3, LastError: "conflict", Metadata: map[string]string{}}, }, expected: CurrentState{Status: entity.RequestStatusLanded, LastError: "", Metadata: map[string]string{"final": "true"}}, }, { name: "same version terminal records uses timestamp tiebreaker", logs: []entity.RequestLog{ - {RequestID: "q/1", TimestampMs: 1000, Status: entity.RequestStatusError, RequestVersion: 3, LastError: "first", Metadata: map[string]string{}}, - {RequestID: "q/1", TimestampMs: 2000, Status: entity.RequestStatusError, RequestVersion: 3, LastError: "second", Metadata: map[string]string{}}, + {RequestID: "q/1", TimestampMs: 1000, Type: entity.RequestLogTypeStatus, Status: entity.RequestStatusError, RequestVersion: 3, LastError: "first", Metadata: map[string]string{}}, + {RequestID: "q/1", TimestampMs: 2000, Type: entity.RequestLogTypeStatus, Status: entity.RequestStatusError, RequestVersion: 3, LastError: "second", Metadata: map[string]string{}}, }, expected: CurrentState{Status: entity.RequestStatusError, LastError: "second", Metadata: map[string]string{}}, }, { name: "terminal status without version is not terminal", logs: []entity.RequestLog{ - {RequestID: "q/1", TimestampMs: 1000, Status: entity.RequestStatusLanded, RequestVersion: 0, LastError: "", Metadata: map[string]string{}}, - {RequestID: "q/1", TimestampMs: 2000, Status: entity.RequestStatusProcessing, RequestVersion: 0, LastError: "", Metadata: map[string]string{"source": "gw"}}, + {RequestID: "q/1", TimestampMs: 1000, Type: entity.RequestLogTypeStatus, Status: entity.RequestStatusLanded, RequestVersion: 0, LastError: "", Metadata: map[string]string{}}, + {RequestID: "q/1", TimestampMs: 2000, Type: entity.RequestLogTypeStatus, Status: entity.RequestStatusSpeculating, RequestVersion: 0, LastError: "", Metadata: map[string]string{"source": "gw"}}, }, - expected: CurrentState{Status: entity.RequestStatusProcessing, LastError: "", Metadata: map[string]string{"source": "gw"}}, + expected: CurrentState{Status: entity.RequestStatusSpeculating, LastError: "", Metadata: map[string]string{"source": "gw"}}, }, { name: "no terminal records falls back to latest timestamp", logs: []entity.RequestLog{ - {RequestID: "q/1", TimestampMs: 1000, Status: entity.RequestStatusStarted, RequestVersion: 1, LastError: "", Metadata: map[string]string{}}, - {RequestID: "q/1", TimestampMs: 3000, Status: entity.RequestStatusValidated, RequestVersion: 2, LastError: "", Metadata: map[string]string{}}, - {RequestID: "q/1", TimestampMs: 2000, Status: entity.RequestStatusProcessing, RequestVersion: 0, LastError: "", Metadata: map[string]string{}}, + {RequestID: "q/1", TimestampMs: 1000, Type: entity.RequestLogTypeStatus, Status: entity.RequestStatusStarted, RequestVersion: 1, LastError: "", Metadata: map[string]string{}}, + {RequestID: "q/1", TimestampMs: 3000, Type: entity.RequestLogTypeStatus, Status: entity.RequestStatusValidated, RequestVersion: 2, LastError: "", Metadata: map[string]string{}}, + {RequestID: "q/1", TimestampMs: 2000, Type: entity.RequestLogTypeStatus, Status: entity.RequestStatusSpeculating, RequestVersion: 0, LastError: "", Metadata: map[string]string{}}, }, expected: CurrentState{Status: entity.RequestStatusValidated, LastError: "", Metadata: map[string]string{}}, }, diff --git a/submitqueue/core/request/terminate.go b/submitqueue/core/request/terminate.go index c4663490..0ba045ca 100644 --- a/submitqueue/core/request/terminate.go +++ b/submitqueue/core/request/terminate.go @@ -140,8 +140,10 @@ func TerminateRequest( logVersion = request.Version } - logEntry := entity.NewRequestLog(request.Queue, requestID, status, logVersion, lastError, metadata) - if err := PublishLog(ctx, registry, logEntry, requestID); err != nil { + logEntry := entity.NewRequestStatusLog(request.Queue, requestID, status, logVersion, lastError, metadata) + // No occurrence: a request reaches a terminal status once, so every repeat + // is a redelivery of that one event and should dedupe away. + if err := PublishLog(ctx, registry, logEntry, requestID, ""); err != nil { return TerminationResult{}, fmt.Errorf("failed to publish request log for %s: %w", requestID, err) } diff --git a/submitqueue/entity/request_log.go b/submitqueue/entity/request_log.go index 343c049d..3206dfb1 100644 --- a/submitqueue/entity/request_log.go +++ b/submitqueue/entity/request_log.go @@ -19,12 +19,16 @@ import ( "time" ) -// RequestLogStatus defines the possible status of a request. Status is customer-friendly and can be displayed to the user. +// RequestStatus is where a request is in the pipeline. It is customer-friendly and can be displayed to the user. // It is different from the request state, which is internal and used to implement a state machine. Request statuses can be // generally added freely by the system without breaking the state machine. // Some statuses correspond to the request state, in which case they should be supplemented with the request state version to be used for reconciliation. // Other statuses are purely informational and can be added freely. // Every status may be accompanied by a last error message and free-formmetadata in the Request Log. It will only be used for display or debugging purposes. +// +// Exactly one status is a request's current position at any moment. Things that merely +// happen to a request while it sits at one — a build starting, a build finishing — are +// not statuses and are not in this vocabulary; they are RequestEvent. type RequestStatus string const ( @@ -47,33 +51,19 @@ const ( // RequestStatusValidated indicates that the request has been validated (duplicate check, merge check etc.) successfully. It corresponds to the RequestStateValidated state. RequestStatusValidated RequestStatus = "validated" - // RequestStatusBatching indicates that the request is waiting to be included in a batch. - RequestStatusBatching RequestStatus = "batching" - // RequestStatusBatched indicates that the request has been included in a new batch and will be sent to speculation. RequestStatusBatched RequestStatus = "batched" - // RequestStatusSpeculating indicates that the request is currently being speculated (e.g., speculative merge/rebase, etc.). + // RequestStatusSpeculating indicates that the batch containing the request has been admitted to speculation: candidate paths are being planned and built. RequestStatusSpeculating RequestStatus = "speculating" - // RequestStatusSpeculated indicates that the request has been successfully speculated and is ready to be validated via a build system. + // RequestStatusSpeculated indicates that the batch containing the request has a build that passed on a path still + // consistent with how its dependencies are resolving, and is waiting for those dependencies to settle before it can land. RequestStatusSpeculated RequestStatus = "speculated" - // RequestStatusBuilding indicates that the request is currently being built (e.g., CI/CD system is building the change on top of the speculation path). - RequestStatusBuilding RequestStatus = "building" - - // RequestStatusBuilt indicates that the request has finished the build step successfully and can move to the next phase, either wait for other requests to finish or move to the land phase. - RequestStatusBuilt RequestStatus = "built" - - // RequestStatusWaitingPath indicates that the request is waiting for other preceiding request in the same speculation path to finish. - RequestStatusWaitingPath RequestStatus = "waitingpath" - // RequestStatusLanding indicates that the request is actively being landed (e.g., source control operation is in progress to push the change to the target branch). RequestStatusLanding RequestStatus = "landing" - // RequestStatusProcessing is the status of a request that is being processed. It corresponds to the RequestStateProcessing state. - RequestStatusProcessing RequestStatus = "processing" - // RequestStatusLanded indicates that the request has been successfully processed and landed. It corresponds to the RequestStateLanded state. RequestStatusLanded RequestStatus = "landed" @@ -91,9 +81,55 @@ const ( RequestStatusCancelled RequestStatus = "cancelled" ) +// RequestEvent is something that happened to a request while it sat at a status, +// rather than a status of its own. +// +// Build progress is what the distinction exists for. A batch funds several +// speculation paths at once and each is built separately, so a build starting or +// finishing says nothing about where the request as a whole is — it is still +// speculating. Were these statuses, one build succeeding while its siblings ran +// would report the request as finished, and go on reporting it that way until the +// batch resolved, because nothing else publishes in between. +// +// Events are not unique per request: each names one build, and a batch may be +// built many times as speculation re-plans. They belong in a request's history +// and are never its current status — which is enforced by the type, since a +// RequestEvent cannot be assigned to RequestSummary.Status. +type RequestEvent string + +const ( + // RequestEventUnknown is the unknown sentinel event. It is set by default when the structure is initialized. It should never be seen in the system. + RequestEventUnknown RequestEvent = "" + + // RequestEventBuilding indicates that one build verifying one speculation path of the batch containing the request has started. + RequestEventBuilding RequestEvent = "building" + + // RequestEventBuilt indicates that one build verifying one speculation path of the batch containing the request finished successfully. + // A build that fails or is cancelled records nothing. + RequestEventBuilt RequestEvent = "built" +) + +// RequestLogType is what a log entry records: the request reaching a status, or +// an event that happened while it was at one. +type RequestLogType string + +const ( + // RequestLogTypeStatus is an entry that records a status. Its Status field is + // set, and it is a candidate for the request's current status. + RequestLogTypeStatus RequestLogType = "status" + + // RequestLogTypeEvent is an entry that records an event. Its Event field is + // set, it carries no request version, and it never becomes a current status. + RequestLogTypeEvent RequestLogType = "event" +) + // RequestLog is an append-only record that captures a point-in-time snapshot of a request's status // for reconciliation purposes. It is stored in a separate database from the request store to support // eventual consistency reconciliation. +// +// An entry records either a status or an event, never both, and Type says which. +// Build the two through NewRequestStatusLog and NewRequestEventLog rather than as +// a literal, so the unset half stays unset. type RequestLog struct { // RequestID is the ID of the request this log entry belongs to. References entity.Request.ID. RequestID string `json:"request_id"` @@ -102,10 +138,15 @@ type RequestLog struct { Queue string `json:"queue"` // TimestampMs is the time this log entry was created, in milliseconds since Unix epoch. TimestampMs int64 `json:"timestamp_ms"` - // Status is the request status at the time this log entry was created. It may contain requests states from the state machine and also display-friendly intermediate statuses. + // Type is what this entry records. An entry read back without one predates the + // field and records a status. + Type RequestLogType `json:"type"` + // Status is the request status this entry records. Set only when Type is RequestLogTypeStatus. Status RequestStatus `json:"status"` + // Event is the event this entry records. Set only when Type is RequestLogTypeEvent. + Event RequestEvent `json:"event"` // RequestVersion is the version of the request at the time this log entry was created. - // Zero if the version is not available. + // Zero if the version is not available, and always zero on an event. RequestVersion int32 `json:"request_version"` // LastError is the last error message associated with the status at the time of this log entry. // Empty string if no error. @@ -115,13 +156,13 @@ type RequestLog struct { Metadata map[string]string `json:"metadata"` } -// NewRequestLog creates a new RequestLog with the given fields. +// NewRequestStatusLog creates a RequestLog recording that a request reached a status. // TimestampMs is set to the current time. If metadata is nil, it will be initialized as an empty map. // queue is the queue processing the request; it scopes requestID, which is only unique within it. // requestVersion is the version of the request entity, should only be set if reporting a request state as a status, otherwise it should be 0. // lastError is the last error message associated with the status at the time of this log entry, empty string if no error. // metadata is a set of key-value pairs providing additional context for this log entry. Not constrained to any specific format or schema, used for display or debugging purposes. -func NewRequestLog(queue string, requestID string, status RequestStatus, requestVersion int32, lastError string, metadata map[string]string) RequestLog { +func NewRequestStatusLog(queue string, requestID string, status RequestStatus, requestVersion int32, lastError string, metadata map[string]string) RequestLog { if metadata == nil { metadata = make(map[string]string) } @@ -129,6 +170,7 @@ func NewRequestLog(queue string, requestID string, status RequestStatus, request RequestID: requestID, Queue: queue, TimestampMs: time.Now().UnixMilli(), + Type: RequestLogTypeStatus, Status: status, RequestVersion: requestVersion, LastError: lastError, @@ -136,6 +178,35 @@ func NewRequestLog(queue string, requestID string, status RequestStatus, request } } +// NewRequestEventLog creates a RequestLog recording that something happened to a +// request while it sat at its current status. +// +// It takes no request version because an event is not a state transition: carrying +// one would let a reader mistake it for a reconcilable status. Arguments otherwise +// match NewRequestStatusLog. +func NewRequestEventLog(queue string, requestID string, event RequestEvent, metadata map[string]string) RequestLog { + if metadata == nil { + metadata = make(map[string]string) + } + return RequestLog{ + RequestID: requestID, + Queue: queue, + TimestampMs: time.Now().UnixMilli(), + Type: RequestLogTypeEvent, + Event: event, + Metadata: metadata, + } +} + +// Value returns what this entry recorded — its status or its event — as the +// single string that identifies it for a message ID, a log line, or display. +func (r RequestLog) Value() string { + if r.Type == RequestLogTypeEvent { + return string(r.Event) + } + return string(r.Status) +} + // ToBytes serializes the RequestLog to JSON bytes for queue message payload. func (r RequestLog) ToBytes() ([]byte, error) { return json.Marshal(r) @@ -143,6 +214,7 @@ func (r RequestLog) ToBytes() ([]byte, error) { // RequestLogFromBytes deserializes a RequestLog from JSON bytes. // If metadata is absent from the JSON, it will be initialized as an empty map. +// An entry without a type predates the field, when every entry recorded a status. func RequestLogFromBytes(data []byte) (RequestLog, error) { var log RequestLog err := json.Unmarshal(data, &log) @@ -152,5 +224,8 @@ func RequestLogFromBytes(data []byte) (RequestLog, error) { if log.Metadata == nil { log.Metadata = make(map[string]string) } + if log.Type == "" { + log.Type = RequestLogTypeStatus + } return log, nil } diff --git a/submitqueue/entity/request_log_test.go b/submitqueue/entity/request_log_test.go index 43f89bbd..3e2f6cc6 100644 --- a/submitqueue/entity/request_log_test.go +++ b/submitqueue/entity/request_log_test.go @@ -22,7 +22,7 @@ import ( ) func TestNewRequestLog_NilMetadata(t *testing.T) { - log := NewRequestLog("queue1", "queue1/100", RequestStatusStarted, 0, "", nil) + log := NewRequestStatusLog("queue1", "queue1/100", RequestStatusStarted, 0, "", nil) assert.NotNil(t, log.Metadata) assert.Empty(t, log.Metadata) @@ -52,7 +52,7 @@ func TestRequestLogFromBytes(t *testing.T) { original := RequestLog{ RequestID: "my-queue/999", TimestampMs: 1709568000000, - Status: RequestStatusProcessing, + Status: RequestStatusSpeculating, RequestVersion: 3, LastError: "timeout", Metadata: map[string]string{"step": "validation", "attempt": "2"}, @@ -104,6 +104,7 @@ func TestRequestLog_SerializationRoundTrip(t *testing.T) { log: RequestLog{ RequestID: "queue1/100", TimestampMs: 1709568000000, + Type: RequestLogTypeStatus, Status: RequestStatusLanded, RequestVersion: 5, LastError: "", @@ -115,6 +116,7 @@ func TestRequestLog_SerializationRoundTrip(t *testing.T) { log: RequestLog{ RequestID: "queue2/200", TimestampMs: 1709568001000, + Type: RequestLogTypeStatus, Status: RequestStatusError, RequestVersion: 2, LastError: "merge conflict detected", @@ -126,6 +128,7 @@ func TestRequestLog_SerializationRoundTrip(t *testing.T) { log: RequestLog{ RequestID: "queue3/300", TimestampMs: 1709568002000, + Type: RequestLogTypeStatus, Status: RequestStatusStarted, RequestVersion: 0, LastError: "", @@ -146,3 +149,61 @@ func TestRequestLog_SerializationRoundTrip(t *testing.T) { }) } } + +// An entry records a status or an event, never both. The constructors are what +// hold that up — the struct can express the invalid combinations, nothing is +// meant to build them — so this pins what each one sets and leaves unset. +// +// The old classifier this replaces asked "is this status really an event?", a +// question the split makes unaskable: an event is a RequestEvent and cannot be +// assigned to Status or to RequestSummary.Status at all. +func TestRequestLogConstructors(t *testing.T) { + t.Run("status entry carries a status and no event", func(t *testing.T) { + log := NewRequestStatusLog("q", "q/1", RequestStatusSpeculating, 4, "boom", map[string]string{"k": "v"}) + + assert.Equal(t, RequestLogTypeStatus, log.Type) + assert.Equal(t, RequestStatusSpeculating, log.Status) + assert.Equal(t, RequestEventUnknown, log.Event) + assert.Equal(t, int32(4), log.RequestVersion) + assert.Equal(t, "boom", log.LastError) + assert.Equal(t, "speculating", log.Value()) + }) + + t.Run("event entry carries an event and no status", func(t *testing.T) { + log := NewRequestEventLog("q", "q/1", RequestEventBuilding, map[string]string{"build_id": "b/7"}) + + assert.Equal(t, RequestLogTypeEvent, log.Type) + assert.Equal(t, RequestEventBuilding, log.Event) + assert.Equal(t, RequestStatusUnknown, log.Status) + assert.Equal(t, "building", log.Value()) + // An event is not a state transition, so it can carry no version for a + // reader to mistake for a reconcilable one. + assert.Zero(t, log.RequestVersion) + }) +} + +// An entry written before the type column existed recorded a status, so it must +// read back as one rather than as the zero type — otherwise every historical +// entry would stop counting towards the current status the moment this shipped. +func TestRequestLogFromBytes_UntypedEntryIsAStatus(t *testing.T) { + legacy := []byte(`{"request_id":"q/1","queue":"q","timestamp_ms":10,"status":"landed","request_version":3}`) + + log, err := RequestLogFromBytes(legacy) + require.NoError(t, err) + + assert.Equal(t, RequestLogTypeStatus, log.Type) + assert.Equal(t, RequestStatusLanded, log.Status) + assert.Equal(t, "landed", log.Value()) +} + +func TestRequestLog_EventSerializationRoundTrip(t *testing.T) { + original := NewRequestEventLog("q", "q/1", RequestEventBuilt, map[string]string{"build_id": "b/7"}) + + data, err := original.ToBytes() + require.NoError(t, err) + + deserialized, err := RequestLogFromBytes(data) + require.NoError(t, err) + + assert.Equal(t, original, deserialized) +} diff --git a/submitqueue/extension/storage/mysql/request_log_store.go b/submitqueue/extension/storage/mysql/request_log_store.go index 6625a6b9..17739a16 100644 --- a/submitqueue/extension/storage/mysql/request_log_store.go +++ b/submitqueue/extension/storage/mysql/request_log_store.go @@ -64,8 +64,8 @@ func (r *requestLogStore) Insert(ctx context.Context, log entity.RequestLog) (re salt := rand.Int64() _, err = r.db.ExecContext(ctx, - "INSERT INTO request_log (queue, request_id, timestamp_ms, salt, status, request_version, last_error, metadata) VALUES (?, ?, ?, ?, ?, ?, ?, ?)", - log.Queue, log.RequestID, log.TimestampMs, salt, log.Status, log.RequestVersion, log.LastError, metadataJSON, + "INSERT INTO request_log (queue, request_id, timestamp_ms, salt, type, status, event, request_version, last_error, metadata) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", + log.Queue, log.RequestID, log.TimestampMs, salt, log.Type, log.Status, log.Event, log.RequestVersion, log.LastError, metadataJSON, ) if err != nil { return fmt.Errorf("failed to insert request log for request_id=%s timestamp_ms=%d: %w", log.RequestID, log.TimestampMs, err) @@ -83,7 +83,7 @@ func (r *requestLogStore) List(ctx context.Context, requestID string) (ret []ent defer func() { op.Complete(retErr) }() rows, err := r.db.QueryContext(ctx, - "SELECT queue, request_id, timestamp_ms, status, request_version, last_error, metadata FROM request_log WHERE queue = ? AND request_id = ? ORDER BY timestamp_ms ASC, salt ASC", + "SELECT queue, request_id, timestamp_ms, type, status, event, request_version, last_error, metadata FROM request_log WHERE queue = ? AND request_id = ? ORDER BY timestamp_ms ASC, salt ASC", r.queue, requestID, ) if err != nil { @@ -96,11 +96,18 @@ func (r *requestLogStore) List(ctx context.Context, requestID string) (ret []ent var log entity.RequestLog var metadataJSON []byte - err := rows.Scan(&log.Queue, &log.RequestID, &log.TimestampMs, &log.Status, &log.RequestVersion, &log.LastError, &metadataJSON) + err := rows.Scan(&log.Queue, &log.RequestID, &log.TimestampMs, &log.Type, &log.Status, &log.Event, &log.RequestVersion, &log.LastError, &metadataJSON) if err != nil { return nil, fmt.Errorf("failed to scan request log row for request_id=%s: %w", requestID, err) } + // A row written before the type column existed carries the column + // default, but a row written by a store that predates it entirely can + // still read back empty; either way it recorded a status. + if log.Type == "" { + log.Type = entity.RequestLogTypeStatus + } + if err := json.Unmarshal(metadataJSON, &log.Metadata); err != nil { return nil, fmt.Errorf("failed to unmarshal metadata for request log request_id=%s: %w", requestID, err) } diff --git a/submitqueue/extension/storage/mysql/request_log_store_test.go b/submitqueue/extension/storage/mysql/request_log_store_test.go index 34aa9c41..9217e076 100644 --- a/submitqueue/extension/storage/mysql/request_log_store_test.go +++ b/submitqueue/extension/storage/mysql/request_log_store_test.go @@ -47,6 +47,7 @@ func TestRequestLogStore_Insert(t *testing.T) { RequestID: "monorepo/1", Queue: testLogQueue, TimestampMs: 1000, + Type: entity.RequestLogTypeStatus, Status: entity.RequestStatusStarted, RequestVersion: 1, LastError: "", @@ -62,7 +63,7 @@ func TestRequestLogStore_Insert(t *testing.T) { name: "success", setup: func(mock sqlmock.Sqlmock) { mock.ExpectExec("INSERT INTO request_log"). - WithArgs(log.Queue, log.RequestID, log.TimestampMs, sqlmock.AnyArg(), log.Status, log.RequestVersion, log.LastError, sqlmock.AnyArg()). + WithArgs(log.Queue, log.RequestID, log.TimestampMs, sqlmock.AnyArg(), log.Type, log.Status, log.Event, log.RequestVersion, log.LastError, sqlmock.AnyArg()). WillReturnResult(sqlmock.NewResult(0, 1)) }, }, @@ -70,7 +71,7 @@ func TestRequestLogStore_Insert(t *testing.T) { name: "exec error", setup: func(mock sqlmock.Sqlmock) { mock.ExpectExec("INSERT INTO request_log"). - WithArgs(log.Queue, log.RequestID, log.TimestampMs, sqlmock.AnyArg(), log.Status, log.RequestVersion, log.LastError, sqlmock.AnyArg()). + WithArgs(log.Queue, log.RequestID, log.TimestampMs, sqlmock.AnyArg(), log.Type, log.Status, log.Event, log.RequestVersion, log.LastError, sqlmock.AnyArg()). WillReturnError(fmt.Errorf("connection reset")) }, wantErr: true, @@ -100,6 +101,7 @@ func TestRequestLogStore_List(t *testing.T) { RequestID: "monorepo/1", Queue: testLogQueue, TimestampMs: 1000, + Type: entity.RequestLogTypeStatus, Status: entity.RequestStatusStarted, RequestVersion: 1, LastError: "", @@ -118,9 +120,9 @@ func TestRequestLogStore_List(t *testing.T) { name: "found", requestID: log.RequestID, setup: func(mock sqlmock.Sqlmock) { - rows := sqlmock.NewRows([]string{"queue", "request_id", "timestamp_ms", "status", "request_version", "last_error", "metadata"}). - AddRow(log.Queue, log.RequestID, log.TimestampMs, string(log.Status), log.RequestVersion, log.LastError, []byte(`{}`)) - mock.ExpectQuery("SELECT queue, request_id, timestamp_ms, status, request_version, last_error, metadata FROM request_log"). + rows := sqlmock.NewRows([]string{"queue", "request_id", "timestamp_ms", "type", "status", "event", "request_version", "last_error", "metadata"}). + AddRow(log.Queue, log.RequestID, log.TimestampMs, string(log.Type), string(log.Status), string(log.Event), log.RequestVersion, log.LastError, []byte(`{}`)) + mock.ExpectQuery("SELECT queue, request_id, timestamp_ms, type, status, event, request_version, last_error, metadata FROM request_log"). WithArgs(testLogQueue, log.RequestID). WillReturnRows(rows) }, @@ -130,8 +132,8 @@ func TestRequestLogStore_List(t *testing.T) { name: "no rows returns ErrNotFound", requestID: "missing", setup: func(mock sqlmock.Sqlmock) { - rows := sqlmock.NewRows([]string{"queue", "request_id", "timestamp_ms", "status", "request_version", "last_error", "metadata"}) - mock.ExpectQuery("SELECT queue, request_id, timestamp_ms, status, request_version, last_error, metadata FROM request_log"). + rows := sqlmock.NewRows([]string{"queue", "request_id", "timestamp_ms", "type", "status", "event", "request_version", "last_error", "metadata"}) + mock.ExpectQuery("SELECT queue, request_id, timestamp_ms, type, status, event, request_version, last_error, metadata FROM request_log"). WithArgs(testLogQueue, "missing"). WillReturnRows(rows) }, @@ -142,7 +144,7 @@ func TestRequestLogStore_List(t *testing.T) { name: "query error", requestID: "bad", setup: func(mock sqlmock.Sqlmock) { - mock.ExpectQuery("SELECT queue, request_id, timestamp_ms, status, request_version, last_error, metadata FROM request_log"). + mock.ExpectQuery("SELECT queue, request_id, timestamp_ms, type, status, event, request_version, last_error, metadata FROM request_log"). WithArgs(testLogQueue, "bad"). WillReturnError(fmt.Errorf("connection reset")) }, diff --git a/submitqueue/extension/storage/mysql/schema/request_log.sql b/submitqueue/extension/storage/mysql/schema/request_log.sql index 261e76cd..d94a22ad 100644 --- a/submitqueue/extension/storage/mysql/schema/request_log.sql +++ b/submitqueue/extension/storage/mysql/schema/request_log.sql @@ -1,12 +1,19 @@ --- request_log is the append-only audit trail of request status transitions. queue leads the +-- request_log is the append-only audit trail of what happened to a request. queue leads the -- PK so the table is shardable by queue; request_id is unique only within its queue. salt -- disambiguates entries sharing a request, queue and millisecond. +-- +-- type says whether a row records a status the request reached or an event that happened +-- while it sat at one, and exactly one of status and event is set accordingly. It defaults +-- to 'status' so rows written before the column existed, when every row was a status, read +-- back correctly. CREATE TABLE IF NOT EXISTS request_log ( queue VARCHAR(255) NOT NULL, request_id VARCHAR(255) NOT NULL, timestamp_ms BIGINT NOT NULL, salt BIGINT NOT NULL, + type VARCHAR(16) NOT NULL DEFAULT 'status', status VARCHAR(64) NOT NULL, + event VARCHAR(64) NOT NULL DEFAULT '', request_version INT NOT NULL, last_error TEXT NOT NULL, metadata JSON NOT NULL, diff --git a/submitqueue/gateway/controller/cancel.go b/submitqueue/gateway/controller/cancel.go index 2c80ef7f..d9db08b4 100644 --- a/submitqueue/gateway/controller/cancel.go +++ b/submitqueue/gateway/controller/cancel.go @@ -110,7 +110,7 @@ func (c *cancelController) Cancel(ctx context.Context, req entity.CancelRequest) if req.Reason != "" { metadata["reason"] = req.Reason } - logEntry := entity.NewRequestLog(req.Queue, req.ID, entity.RequestStatusCancelling, 0, "", metadata) + logEntry := entity.NewRequestStatusLog(req.Queue, req.ID, entity.RequestStatusCancelling, 0, "", metadata) if err := c.materializer.PersistLog(ctx, logEntry); err != nil { return fmt.Errorf("failed to insert cancelling log for sqid=%s: %w", req.ID, err) } diff --git a/submitqueue/gateway/controller/land.go b/submitqueue/gateway/controller/land.go index 588376d4..0f766b48 100644 --- a/submitqueue/gateway/controller/land.go +++ b/submitqueue/gateway/controller/land.go @@ -165,6 +165,7 @@ func (c *landController) Land(ctx context.Context, req entity.LandRequest) (resu RequestID: req.ID, Queue: req.Queue, TimestampMs: receivedAtMs, + Type: entity.RequestLogTypeStatus, Status: entity.RequestStatusAccepted, Metadata: map[string]string{}, } diff --git a/submitqueue/gateway/controller/land_test.go b/submitqueue/gateway/controller/land_test.go index de7b2224..c957ef23 100644 --- a/submitqueue/gateway/controller/land_test.go +++ b/submitqueue/gateway/controller/land_test.go @@ -434,6 +434,7 @@ func TestLand_PublishesToQueue(t *testing.T) { RequestID: "test-queue/123", Queue: "test-queue", TimestampMs: receiptSummary.ReceivedAtMs, + Type: entity.RequestLogTypeStatus, Status: entity.RequestStatusAccepted, Metadata: map[string]string{}, }, persistedLog) diff --git a/submitqueue/gateway/controller/log/log_test.go b/submitqueue/gateway/controller/log/log_test.go index 0cf73deb..d943e70e 100644 --- a/submitqueue/gateway/controller/log/log_test.go +++ b/submitqueue/gateway/controller/log/log_test.go @@ -154,6 +154,6 @@ func newLogControllerStore(ctrl *gomock.Controller, insertErr, getErr, updateErr } func newRequestLog(requestID string, status entity.RequestStatus, requestVersion int32, lastError string, metadata map[string]string) *entity.RequestLog { - log := entity.NewRequestLog("test-queue", requestID, status, requestVersion, lastError, metadata) + log := entity.NewRequestStatusLog("test-queue", requestID, status, requestVersion, lastError, metadata) return &log } diff --git a/submitqueue/orchestrator/controller/batch/batch.go b/submitqueue/orchestrator/controller/batch/batch.go index 640fee04..f720ec29 100644 --- a/submitqueue/orchestrator/controller/batch/batch.go +++ b/submitqueue/orchestrator/controller/batch/batch.go @@ -310,14 +310,16 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er // Record the "batched" status in the request log. This status corresponds to // the RequestStateBatched transition CAS'd above, so it carries the request - // version for reconciliation. The message ID is scoped to (requestID, status), - // so a redelivery that creates a fresh batch re-emits "batched" with a - // different batch_id but is deduped to the first entry — acceptable, the - // request is batched either way. - logEntry := entity.NewRequestLog(request.Queue, request.ID, entity.RequestStatusBatched, request.Version, "", map[string]string{ + // version for reconciliation. No occurrence is passed, which scopes the + // message ID to (requestID, status): a redelivery that creates a fresh batch + // re-emits "batched" with a different batch_id but is deduped to the first + // entry — acceptable, the request is batched either way, and passing the + // batch ID here would instead surface every abandoned attempt as its own + // entry. + logEntry := entity.NewRequestStatusLog(request.Queue, request.ID, entity.RequestStatusBatched, request.Version, "", map[string]string{ "batch_id": batch.ID, }) - if err := corerequest.PublishLog(ctx, c.registry, logEntry, request.ID); err != nil { + if err := corerequest.PublishLog(ctx, c.registry, logEntry, request.ID, ""); err != nil { metrics.NamedCounter(c.metricsScope, opName, "request_log_errors", 1) metrics.NamedCounter(c.metricsScope, opName, "batch_abandoned_created", 1) return fmt.Errorf("failed to publish request log for request %s: %w", request.ID, err) diff --git a/submitqueue/orchestrator/controller/buildsignal/BUILD.bazel b/submitqueue/orchestrator/controller/buildsignal/BUILD.bazel index acec8611..72e4ee7f 100644 --- a/submitqueue/orchestrator/controller/buildsignal/BUILD.bazel +++ b/submitqueue/orchestrator/controller/buildsignal/BUILD.bazel @@ -9,6 +9,7 @@ go_library( "//platform/consumer:go_default_library", "//platform/metrics:go_default_library", "//submitqueue/core/publish:go_default_library", + "//submitqueue/core/request:go_default_library", "//submitqueue/core/topickey:go_default_library", "//submitqueue/entity:go_default_library", "//submitqueue/extension/buildrunner:go_default_library", diff --git a/submitqueue/orchestrator/controller/buildsignal/buildsignal.go b/submitqueue/orchestrator/controller/buildsignal/buildsignal.go index 05a4e49f..2663c508 100644 --- a/submitqueue/orchestrator/controller/buildsignal/buildsignal.go +++ b/submitqueue/orchestrator/controller/buildsignal/buildsignal.go @@ -43,11 +43,13 @@ import ( "context" "errors" "fmt" + "strconv" "github.com/uber-go/tally" "github.com/uber/submitqueue/platform/consumer" "github.com/uber/submitqueue/platform/metrics" "github.com/uber/submitqueue/submitqueue/core/publish" + corerequest "github.com/uber/submitqueue/submitqueue/core/request" "github.com/uber/submitqueue/submitqueue/core/topickey" "github.com/uber/submitqueue/submitqueue/entity" "github.com/uber/submitqueue/submitqueue/extension/buildrunner" @@ -186,7 +188,7 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er return fmt.Errorf("failed to build runner for batch %s: %w", batch.ID, err) } - status, _, err := buildRunner.Status(ctx, buildID) + status, buildMeta, err := buildRunner.Status(ctx, buildID) if err != nil { metrics.NamedCounter(c.metricsScope, opName, "status_errors", 1) return fmt.Errorf("failed to get status for build %s: %w", buildID.ID, err) @@ -222,6 +224,16 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er } if status != build.Status { + // Tell the members what their build is doing, before the record moves. + // This branch is the only place a change of status is observed, so a log + // published after the write would be lost for good when it failed: the + // replay would read the already-updated record and take neither branch. + // Publishing first means the replay re-publishes and the queue dedupes. + if err := c.publishBuildLogs(ctx, batch, build, status, buildMeta); err != nil { + metrics.NamedCounter(c.metricsScope, opName, "request_log_errors", 1) + return fmt.Errorf("failed to publish request logs for build %s: %w", build.ID, err) + } + build.Status = status if err := store.GetBuildStore().Update(ctx, build); err != nil { metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1) @@ -272,6 +284,58 @@ func pollDelay(status entity.BuildStatus) int64 { } } +// publishBuildLogs records this build's progress against every request in the +// batch it verifies. +// +// Both entries are events rather than positions. A head funds several paths and +// each is built separately, so one build starting or finishing says nothing +// about where the request as a whole is — it is still speculating — and letting +// either become the current status would report a request as finished while its +// sibling builds were still running (see entity.IsRequestStatusEvent). What they +// carry instead is identity: which path, which attempt, and the runner's URL, so +// a request stuck in a rebuild loop shows a pair per attempt rather than one +// indistinguishable entry. +// +// The build ID is the occurrence, so however many transitions a build passes +// through and however often its delivery is replayed, each entry lands once per +// build. +func (c *Controller) publishBuildLogs( + ctx context.Context, + batch entity.Batch, + build entity.Build, + status entity.BuildStatus, + buildMeta entity.BuildMetadata, +) error { + logMeta := map[string]string{ + "batch_id": batch.ID, + "build_id": build.ID, + "path_id": build.PathID, + "attempt": strconv.Itoa(build.Attempt), + } + // Runners report the run's web address under "url"; it is the only part of + // this that a customer can click, so it is worth carrying when present. + if url := buildMeta["url"]; url != "" { + logMeta["build_url"] = url + } + + // Emitted on any observed transition, not just the first, so a build that + // jumps straight from Accepted to a terminal status is still reported as + // having run. Repeats collapse on the occurrence. + if err := corerequest.PublishBatchEvents(ctx, c.registry, batch.Queue, batch.Contains, + entity.RequestEventBuilding, build.ID, logMeta); err != nil { + return err + } + + // Only success is built. A build that failed or was cancelled has no + // completion to report — its path is refuted, and what happens next is the + // speculate run's decision, not this one's. + if status != entity.BuildStatusSucceeded { + return nil + } + return corerequest.PublishBatchEvents(ctx, c.registry, batch.Queue, batch.Contains, + entity.RequestEventBuilt, build.ID, logMeta) +} + // unwanted reports whether nothing wants this build running any more: its // batch has halted, its path was called off or has moved to another attempt, // or the attempt's link names a different build (this one lost a dispatch diff --git a/submitqueue/orchestrator/controller/buildsignal/buildsignal_test.go b/submitqueue/orchestrator/controller/buildsignal/buildsignal_test.go index e320849b..6965d382 100644 --- a/submitqueue/orchestrator/controller/buildsignal/buildsignal_test.go +++ b/submitqueue/orchestrator/controller/buildsignal/buildsignal_test.go @@ -17,6 +17,7 @@ package buildsignal import ( "context" "errors" + "fmt" "testing" "github.com/stretchr/testify/assert" @@ -56,6 +57,9 @@ type testHarness struct { pathBuilds *storagemock.MockPathBuildStore signalPub *queuemock.MockPublisher speculatePub *queuemock.MockPublisher + // logs holds the request-log entries this poll reported to the members of + // the batch the build verifies. + logs []entity.RequestLog } // staticStorageFactory resolves every queue to one fixed store aggregate. @@ -64,7 +68,13 @@ type staticStorageFactory struct{ store storage.Storage } // For returns the fixed store aggregate for any queue. func (f staticStorageFactory) For(storage.Config) (storage.Storage, error) { return f.store, nil } +// testRequestID is the one member of the batch under test, so the request-log +// fan-out has somebody to report to. +const testRequestID = "test-queue/1" + func newTestHarness(t *testing.T, ctrl *gomock.Controller, batchState entity.BatchState) *testHarness { + h := &testHarness{} + br := buildrunnermock.NewMockBuildRunner(ctrl) brFactory := buildrunnermock.NewMockFactory(ctrl) brFactory.EXPECT().For(gomock.Any()).Return(br, nil).AnyTimes() @@ -77,16 +87,30 @@ func newTestHarness(t *testing.T, ctrl *gomock.Controller, batchState entity.Bat speculateQ := queuemock.NewMockQueue(ctrl) speculateQ.EXPECT().Publisher().Return(speculatePub).AnyTimes() + logPub := queuemock.NewMockPublisher(ctrl) + logPub.EXPECT().Publish(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn( + func(_ context.Context, _ string, msg entityqueue.Message) error { + entry, err := entity.RequestLogFromBytes(msg.Payload) + require.NoError(t, err) + h.logs = append(h.logs, entry) + return nil + }, + ).AnyTimes() + logQ := queuemock.NewMockQueue(ctrl) + logQ.EXPECT().Publisher().Return(logPub).AnyTimes() + registry, err := consumer.NewTopicRegistry([]consumer.TopicConfig{ {Key: topickey.TopicKeyBuildSignal, Name: "buildsignal", Queue: signalQ}, {Key: topickey.TopicKeySpeculate, Name: "speculate", Queue: speculateQ}, + {Key: topickey.TopicKeyLog, Name: "log", Queue: logQ}, }) require.NoError(t, err) builds := storagemock.NewMockBuildStore(ctrl) batchStore := storagemock.NewMockBatchStore(ctrl) batchStore.EXPECT().Get(gomock.Any(), testBatchID).Return(entity.Batch{ - ID: testBatchID, Queue: "test-queue", State: batchState, Version: 1, + ID: testBatchID, Queue: "test-queue", Contains: []string{testRequestID}, + State: batchState, Version: 1, }, nil).AnyTimes() // The path set and link stores are wired read-only: their getters answer, @@ -110,16 +134,16 @@ func newTestHarness(t *testing.T, ctrl *gomock.Controller, batchState entity.Bat topickey.TopicKeyBuildSignal, "orchestrator-buildsignal", ) - return &testHarness{ - controller: c, - br: br, - builds: builds, - batchStore: batchStore, - pathSets: pathSets, - pathBuilds: pathBuilds, - signalPub: signalPub, - speculatePub: speculatePub, - } + // Populated rather than replaced: the log publisher above closes over h. + h.controller = c + h.br = br + h.builds = builds + h.batchStore = batchStore + h.pathSets = pathSets + h.pathBuilds = pathBuilds + h.signalPub = signalPub + h.speculatePub = speculatePub + return h } // wanted wires the kill-list reads to say the build is still wanted: its entry @@ -570,3 +594,130 @@ func TestProcess_Errors(t *testing.T) { }) } } + +// Build progress is reported to the members of the batch the build verifies, +// carrying the identity a customer can act on — which path, which attempt, and +// the runner's URL. +// +// "built" is success only. A head funds several paths at once, so a failed or +// cancelled build is not the request finishing anything: its path is refuted +// and what happens next is the speculate run's decision. +func TestProcess_ReportsBuildProgressToMembers(t *testing.T) { + tests := []struct { + name string + status entity.BuildStatus + wantEvents []entity.RequestEvent + }{ + { + name: "running is building only", + status: entity.BuildStatusRunning, + wantEvents: []entity.RequestEvent{entity.RequestEventBuilding}, + }, + { + name: "success is building then built", + status: entity.BuildStatusSucceeded, + wantEvents: []entity.RequestEvent{entity.RequestEventBuilding, entity.RequestEventBuilt}, + }, + { + name: "failure reports no completion", + status: entity.BuildStatusFailed, + wantEvents: []entity.RequestEvent{entity.RequestEventBuilding}, + }, + { + name: "cancellation reports no completion", + status: entity.BuildStatusCancelled, + wantEvents: []entity.RequestEvent{entity.RequestEventBuilding}, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + ctrl := gomock.NewController(t) + h := newTestHarness(t, ctrl, entity.BatchStateSpeculating) + + h.builds.EXPECT().Get(gomock.Any(), testBuildID).Return(testBuild(entity.BuildStatusAccepted), nil) + h.br.EXPECT().Status(gomock.Any(), gomock.Any()).Return( + tt.status, entity.BuildMetadata{"url": "https://ci.example.com/builds/7"}, nil) + h.builds.EXPECT().Update(gomock.Any(), testBuild(tt.status)).Return(nil) + h.speculatePub.EXPECT().Publish(gomock.Any(), "speculate", gomock.Any()).Return(nil) + if !tt.status.IsTerminal() { + h.wanted() + delivery := delivery(t, ctrl) + delivery.EXPECT().Hold(gomock.Any()) + require.NoError(t, h.controller.Process(context.Background(), delivery)) + } else { + require.NoError(t, h.controller.Process(context.Background(), delivery(t, ctrl))) + } + + require.Len(t, h.logs, len(tt.wantEvents)) + for i, want := range tt.wantEvents { + assert.Equal(t, entity.RequestLogTypeEvent, h.logs[i].Type) + assert.Equal(t, want, h.logs[i].Event) + // Nothing this stage records is a status, so the summary has + // nothing here it could sit on. + assert.Equal(t, entity.RequestStatusUnknown, h.logs[i].Status) + assert.Equal(t, testRequestID, h.logs[i].RequestID) + assert.Equal(t, testBuildID, h.logs[i].Metadata["build_id"]) + assert.Equal(t, testPathID, h.logs[i].Metadata["path_id"]) + assert.Equal(t, "https://ci.example.com/builds/7", h.logs[i].Metadata["build_url"]) + // Zero version: an event is not a transition of the request's + // own state machine, and carrying a version would let the + // materializer mistake it for one. + assert.Zero(t, h.logs[i].RequestVersion) + } + }) + } +} + +// A poll that sees no change reports nothing. The entries are keyed on the +// build, so a repeat would dedupe anyway, but publishing one per poll would put +// a queue write behind every two-second tick of every live build. +func TestProcess_UnchangedStatusReportsNothing(t *testing.T) { + ctrl := gomock.NewController(t) + h := newTestHarness(t, ctrl, entity.BatchStateSpeculating) + + h.builds.EXPECT().Get(gomock.Any(), testBuildID).Return(testBuild(entity.BuildStatusRunning), nil) + h.br.EXPECT().Status(gomock.Any(), gomock.Any()).Return(entity.BuildStatusRunning, nil, nil) + h.wanted() + h.speculatePub.EXPECT().Publish(gomock.Any(), "speculate", gomock.Any()).Return(nil) + + d := delivery(t, ctrl) + d.EXPECT().Hold(gomock.Any()) + require.NoError(t, h.controller.Process(context.Background(), d)) + + assert.Empty(t, h.logs) +} + +// The report goes out before the build record moves. This branch is the only +// place a change of status is seen, so a report written after the record would +// be lost for good if it failed — the replay would read the updated record and +// take neither branch. +func TestProcess_ReportsBeforeRecordingStatus(t *testing.T) { + ctrl := gomock.NewController(t) + h := newTestHarness(t, ctrl, entity.BatchStateSpeculating) + + h.builds.EXPECT().Get(gomock.Any(), testBuildID).Return(testBuild(entity.BuildStatusAccepted), nil) + h.br.EXPECT().Status(gomock.Any(), gomock.Any()).Return(entity.BuildStatusSucceeded, nil, nil) + + // No Update and no speculate publish expectation: gomock fails the test if + // the run gets past the failed report. + h.controller.registry = failingLogRegistry(t, ctrl) + + require.Error(t, h.controller.Process(context.Background(), delivery(t, ctrl))) +} + +// failingLogRegistry answers every topic with a publisher that refuses. +func failingLogRegistry(t *testing.T, ctrl *gomock.Controller) consumer.TopicRegistry { + t.Helper() + pub := queuemock.NewMockPublisher(ctrl) + pub.EXPECT().Publish(gomock.Any(), gomock.Any(), gomock.Any()). + Return(fmt.Errorf("enqueue failed")).AnyTimes() + q := queuemock.NewMockQueue(ctrl) + q.EXPECT().Publisher().Return(pub).AnyTimes() + registry, err := consumer.NewTopicRegistry([]consumer.TopicConfig{ + {Key: topickey.TopicKeyLog, Name: "log", Queue: q}, + {Key: topickey.TopicKeySpeculate, Name: "speculate", Queue: q}, + }) + require.NoError(t, err) + return registry +} diff --git a/submitqueue/orchestrator/controller/merge/BUILD.bazel b/submitqueue/orchestrator/controller/merge/BUILD.bazel index 61aceedc..6370a4fc 100644 --- a/submitqueue/orchestrator/controller/merge/BUILD.bazel +++ b/submitqueue/orchestrator/controller/merge/BUILD.bazel @@ -13,6 +13,7 @@ go_library( "//platform/base/messagequeue:go_default_library", "//platform/consumer:go_default_library", "//platform/metrics:go_default_library", + "//submitqueue/core/request:go_default_library", "//submitqueue/entity:go_default_library", "//submitqueue/extension/storage:go_default_library", "@com_github_uber_go_tally//:go_default_library", diff --git a/submitqueue/orchestrator/controller/merge/merge.go b/submitqueue/orchestrator/controller/merge/merge.go index c2d5d8b1..0fa9c042 100644 --- a/submitqueue/orchestrator/controller/merge/merge.go +++ b/submitqueue/orchestrator/controller/merge/merge.go @@ -35,6 +35,7 @@ import ( entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" "github.com/uber/submitqueue/platform/consumer" "github.com/uber/submitqueue/platform/metrics" + corerequest "github.com/uber/submitqueue/submitqueue/core/request" "github.com/uber/submitqueue/submitqueue/entity" "github.com/uber/submitqueue/submitqueue/extension/storage" ) @@ -153,6 +154,16 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er return fmt.Errorf("failed to build merge request for batch %s: %w", batch.ID, err) } + // Report that the members are landing before the request goes out, so a + // publish that fails nacks with nothing announced, and a crash between the + // two re-publishes under the same occurrence and dedupes. + if err := corerequest.PublishBatchLogs(ctx, c.registry, batch.Queue, batch.Contains, + entity.RequestStatusLanding, batch.ID, map[string]string{"batch_id": batch.ID}, + ); err != nil { + metrics.NamedCounter(c.metricsScope, opName, "request_log_errors", 1) + return fmt.Errorf("failed to publish request log for batch %s: %w", batch.ID, err) + } + if err := c.publish(ctx, c.runwayTopicKey, req, batch.Queue); err != nil { metrics.NamedCounter(c.metricsScope, opName, "publish_errors", 1) return fmt.Errorf("failed to publish to runway merge: %w", err) diff --git a/submitqueue/orchestrator/controller/merge/merge_test.go b/submitqueue/orchestrator/controller/merge/merge_test.go index 92a3fb1b..90c63cd6 100644 --- a/submitqueue/orchestrator/controller/merge/merge_test.go +++ b/submitqueue/orchestrator/controller/merge/merge_test.go @@ -73,6 +73,43 @@ func newController(t *testing.T, store *storagemock.MockStorage, registry consum ) } +// publishes records what a controller run published, in order and by topic. The +// controller writes to two topics now — the request-log fan-out and the runway +// merge request — so tests need to tell them apart and to see which came first. +type publishes struct { + inOrder []string + byTopic map[string][]entityqueue.Message +} + +// newRegistry builds a registry carrying both topics this controller publishes +// to, recording every message. failTopic names a topic whose publishes fail; +// empty means they all succeed. +func newRegistry(t *testing.T, ctrl *gomock.Controller, failTopic string) (consumer.TopicRegistry, *publishes) { + rec := &publishes{byTopic: map[string][]entityqueue.Message{}} + + pub := queuemock.NewMockPublisher(ctrl) + pub.EXPECT().Publish(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn( + func(_ context.Context, topic string, msg entityqueue.Message) error { + if topic == failTopic { + return fmt.Errorf("enqueue failed") + } + rec.inOrder = append(rec.inOrder, topic) + rec.byTopic[topic] = append(rec.byTopic[topic], msg) + return nil + }, + ).AnyTimes() + + q := queuemock.NewMockQueue(ctrl) + q.EXPECT().Publisher().Return(pub).AnyTimes() + + registry, err := consumer.NewTopicRegistry([]consumer.TopicConfig{ + {Key: runwaymq.TopicKeyMerge, Name: "runway-merge", Queue: q}, + {Key: topickey.TopicKeyLog, Name: "log", Queue: q}, + }) + require.NoError(t, err) + return registry, rec +} + func TestNewController(t *testing.T) { ctrl := gomock.NewController(t) store := storagemock.NewMockStorage(ctrl) @@ -130,16 +167,20 @@ func TestProcess_PublishesFullPayloadToRunway(t *testing.T) { pub := queuemock.NewMockPublisher(ctrl) pub.EXPECT().Publish(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn( func(_ context.Context, topic string, msg entityqueue.Message) error { + if topic != "runway-merge" { + return nil + } gotTopic = topic gotPayload = msg.Payload return nil }, - ) + ).AnyTimes() q := queuemock.NewMockQueue(ctrl) q.EXPECT().Publisher().Return(pub).AnyTimes() - registry, err := consumer.NewTopicRegistry( - []consumer.TopicConfig{{Key: runwaymq.TopicKeyMerge, Name: "runway-merge", Queue: q}}, - ) + registry, err := consumer.NewTopicRegistry([]consumer.TopicConfig{ + {Key: runwaymq.TopicKeyMerge, Name: "runway-merge", Queue: q}, + {Key: topickey.TopicKeyLog, Name: "log", Queue: q}, + }) require.NoError(t, err) c := newController(t, store, registry) @@ -179,53 +220,107 @@ func TestProcess_HaltedBatchSkips(t *testing.T) { batchStore := storagemock.NewMockBatchStore(ctrl) batchStore.EXPECT().Get(gomock.Any(), batchID).Return(batch, nil) - // No request-store reads and no publish for a halted batch: gomock - // fails if GetRequestStore or Publish is touched. + // No request-store reads and no publish for a halted batch: the + // members are told nothing and runway is not asked to merge. store := storagemock.NewMockStorage(ctrl) store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes() - pub := queuemock.NewMockPublisher(ctrl) - q := queuemock.NewMockQueue(ctrl) - q.EXPECT().Publisher().Return(pub).AnyTimes() - registry, err := consumer.NewTopicRegistry( - []consumer.TopicConfig{{Key: runwaymq.TopicKeyMerge, Name: "runway-merge", Queue: q}}, - ) - require.NoError(t, err) + registry, rec := newRegistry(t, ctrl, "") c := newController(t, store, registry) require.NoError(t, c.Process(context.Background(), newDelivery(t, ctrl, batchID, batch.Queue))) + assert.Empty(t, rec.inOrder) }) } } -func TestProcess_PublishFailureReturnsError(t *testing.T) { +// TestProcess_ReportsLandingBeforeDispatch covers the request-log half of this +// stage: every member of the batch is told it is landing, and it is told before +// the merge request goes out. The ordering is what makes a lost log entry +// recoverable — a failure here nacks with nothing announced to runway, and the +// redelivery re-publishes under the same occurrence, which the queue dedupes. +func TestProcess_ReportsLandingBeforeDispatch(t *testing.T) { ctrl := gomock.NewController(t) - const batchID = "test-queue/batch/2" - req := entity.Request{ID: "test-queue/1", Queue: "test-queue", LandStrategy: mergestrategy.MergeStrategyRebase} - batch := entity.Batch{ID: batchID, Queue: "test-queue", Contains: []string{req.ID}, State: entity.BatchStateMerging, Version: 1} + const batchID = "test-queue/batch/3" + req1 := entity.Request{ID: "test-queue/1", Queue: "test-queue", LandStrategy: mergestrategy.MergeStrategyRebase} + req2 := entity.Request{ID: "test-queue/2", Queue: "test-queue", LandStrategy: mergestrategy.MergeStrategyRebase} + batch := entity.Batch{ + ID: batchID, Queue: "test-queue", + Contains: []string{req1.ID, req2.ID}, + State: entity.BatchStateMerging, Version: 2, + } batchStore := storagemock.NewMockBatchStore(ctrl) batchStore.EXPECT().Get(gomock.Any(), batchID).Return(batch, nil) reqStore := storagemock.NewMockRequestStore(ctrl) - reqStore.EXPECT().Get(gomock.Any(), req.ID).Return(req, nil) + reqStore.EXPECT().Get(gomock.Any(), req1.ID).Return(req1, nil) + reqStore.EXPECT().Get(gomock.Any(), req2.ID).Return(req2, nil) store := storagemock.NewMockStorage(ctrl) store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes() store.EXPECT().GetRequestStore().Return(reqStore).AnyTimes() - pub := queuemock.NewMockPublisher(ctrl) - pub.EXPECT().Publish(gomock.Any(), gomock.Any(), gomock.Any()).Return(fmt.Errorf("enqueue failed")) - q := queuemock.NewMockQueue(ctrl) - q.EXPECT().Publisher().Return(pub).AnyTimes() - registry, err := consumer.NewTopicRegistry( - []consumer.TopicConfig{{Key: runwaymq.TopicKeyMerge, Name: "runway-merge", Queue: q}}, - ) - require.NoError(t, err) - + registry, rec := newRegistry(t, ctrl, "") c := newController(t, store, registry) - err = c.Process(context.Background(), newDelivery(t, ctrl, batchID, batch.Queue)) - require.Error(t, err) + require.NoError(t, c.Process(context.Background(), newDelivery(t, ctrl, batchID, batch.Queue))) + + // One entry per member, then the dispatch. + assert.Equal(t, []string{"log", "log", "runway-merge"}, rec.inOrder) + + logs := rec.byTopic["log"] + require.Len(t, logs, 2) + for i, requestID := range []string{req1.ID, req2.ID} { + entry, err := entity.RequestLogFromBytes(logs[i].Payload) + require.NoError(t, err) + assert.Equal(t, requestID, entry.RequestID) + assert.Equal(t, entity.RequestStatusLanding, entry.Status) + assert.Equal(t, batch.ID, entry.Metadata["batch_id"]) + // Partitioned per request so its entries stay ordered, and scoped to the + // batch so a redelivery of this message dedupes. + assert.Equal(t, requestID, logs[i].PartitionKey) + assert.Equal(t, requestID+"/landing/"+batch.ID, logs[i].ID) + } +} + +func TestProcess_PublishFailureReturnsError(t *testing.T) { + tests := []struct { + name string + failTopic string + }{ + {name: "runway dispatch fails", failTopic: "runway-merge"}, + {name: "request log fails", failTopic: "log"}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + ctrl := gomock.NewController(t) + + const batchID = "test-queue/batch/2" + req := entity.Request{ID: "test-queue/1", Queue: "test-queue", LandStrategy: mergestrategy.MergeStrategyRebase} + batch := entity.Batch{ID: batchID, Queue: "test-queue", Contains: []string{req.ID}, State: entity.BatchStateMerging, Version: 1} + + batchStore := storagemock.NewMockBatchStore(ctrl) + batchStore.EXPECT().Get(gomock.Any(), batchID).Return(batch, nil) + reqStore := storagemock.NewMockRequestStore(ctrl) + reqStore.EXPECT().Get(gomock.Any(), req.ID).Return(req, nil).AnyTimes() + + store := storagemock.NewMockStorage(ctrl) + store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes() + store.EXPECT().GetRequestStore().Return(reqStore).AnyTimes() + + registry, rec := newRegistry(t, ctrl, tt.failTopic) + + c := newController(t, store, registry) + require.Error(t, c.Process(context.Background(), newDelivery(t, ctrl, batchID, batch.Queue))) + + // A failed log publish must stop the run before runway hears about + // the merge, so the redelivery can repair the log entry. + if tt.failTopic == "log" { + assert.Empty(t, rec.byTopic["runway-merge"]) + } + }) + } } func TestProcess_BatchStoreGetFailureNotRetryable(t *testing.T) { diff --git a/submitqueue/orchestrator/controller/mergeconflictsignal/mergeconflictsignal.go b/submitqueue/orchestrator/controller/mergeconflictsignal/mergeconflictsignal.go index 4bb4315b..2402f601 100644 --- a/submitqueue/orchestrator/controller/mergeconflictsignal/mergeconflictsignal.go +++ b/submitqueue/orchestrator/controller/mergeconflictsignal/mergeconflictsignal.go @@ -141,8 +141,8 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er } request.Version = newVersion - logEntry := entity.NewRequestLog(request.Queue, request.ID, entity.RequestStatusValidated, request.Version, "", nil) - if err := corerequest.PublishLog(ctx, c.registry, logEntry, request.ID); err != nil { + logEntry := entity.NewRequestStatusLog(request.Queue, request.ID, entity.RequestStatusValidated, request.Version, "", nil) + if err := corerequest.PublishLog(ctx, c.registry, logEntry, request.ID, ""); err != nil { metrics.NamedCounter(c.metricsScope, opName, "log_errors", 1) return fmt.Errorf("failed to publish request log for %s: %w", request.ID, err) } @@ -188,8 +188,8 @@ func (c *Controller) failRequest(ctx context.Context, store storage.Storage, req request.Version = newVersion } - logEntry := entity.NewRequestLog(request.Queue, request.ID, entity.RequestStatusError, request.Version, reason, nil) - if err := corerequest.PublishLog(ctx, c.registry, logEntry, request.ID); err != nil { + logEntry := entity.NewRequestStatusLog(request.Queue, request.ID, entity.RequestStatusError, request.Version, reason, nil) + if err := corerequest.PublishLog(ctx, c.registry, logEntry, request.ID, ""); err != nil { return fmt.Errorf("failed to publish request log for %s: %w", request.ID, err) } return nil diff --git a/submitqueue/orchestrator/controller/speculate/BUILD.bazel b/submitqueue/orchestrator/controller/speculate/BUILD.bazel index fdcf5314..fe47b2c8 100644 --- a/submitqueue/orchestrator/controller/speculate/BUILD.bazel +++ b/submitqueue/orchestrator/controller/speculate/BUILD.bazel @@ -21,6 +21,7 @@ go_library( "//platform/metrics:go_default_library", "//submitqueue/core/batch:go_default_library", "//submitqueue/core/publish:go_default_library", + "//submitqueue/core/request:go_default_library", "//submitqueue/core/topickey:go_default_library", "//submitqueue/entity:go_default_library", "//submitqueue/extension/speculation/speculator:go_default_library", diff --git a/submitqueue/orchestrator/controller/speculate/finalize.go b/submitqueue/orchestrator/controller/speculate/finalize.go index 206be545..04cd5256 100644 --- a/submitqueue/orchestrator/controller/speculate/finalize.go +++ b/submitqueue/orchestrator/controller/speculate/finalize.go @@ -22,6 +22,7 @@ import ( "github.com/uber/submitqueue/platform/metrics" corebatch "github.com/uber/submitqueue/submitqueue/core/batch" + corerequest "github.com/uber/submitqueue/submitqueue/core/request" "github.com/uber/submitqueue/submitqueue/core/topickey" "github.com/uber/submitqueue/submitqueue/entity" "github.com/uber/submitqueue/submitqueue/extension/storage" @@ -75,11 +76,16 @@ func (c *Controller) finalize(ctx context.Context, snap *snapshot) error { // Fold in what the facts already imply about this head's paths: // a finished dependency breaks every path that bet against it. set, exists := snap.pathSets[batch.ID] + before, hadPassed := passedEntry(set) if exists && cancelBrokenPathsInSet(&set, *snap, nowMs) { snap.pathSets[batch.ID] = set snap.markDirty(batch.ID) } + if err := c.reportSpeculation(ctx, batch, set, *snap, before, hadPassed); err != nil { + return err + } + decision := decide(batch, set, *snap) if decision == outcomeWait { stillOpen = append(stillOpen, batch) @@ -122,6 +128,66 @@ func (c *Controller) finalize(ctx context.Context, snap *snapshot) error { return nil } +// reportSpeculation tells a head's members how far speculation has got. +// +// Two moments are worth reporting and neither is a batch state — a head is +// BatchStateSpeculating from admission until its outcome, so the request log is +// the only place either becomes visible: +// +// - the head has a live passed path. Its own work is done and what remains is +// other batches finishing, a wait that can run for minutes and reads very +// differently to still building. +// - it just lost the one it had, because a dependency resolved against that +// path's guess. The head is back to building, and without this its members +// would go on reading as speculated through the whole rebuild. +// +// The second is why before is taken from passedEntry rather than livePassedPath: +// both that predicate and the fold above exclude a contradicted path, so a pair +// of livePassedPath calls could never see the loss happen. What is compared is +// "held a passed build" before the fold against "still has one worth waiting on" +// after it, and only the run that does the breaking sees the difference — every +// later run finds the entry already cancelled. +// +// Both facts are derived from the snapshot rather than stored, so this runs on +// every pass over an open head and relies on the occurrence to collapse the +// repeats: a path ID hashes its head along with its assumptions, so it names the +// batch too, and one passed path re-observed by a hundred runs is a single entry +// while a different path winning after a re-plan is correctly a new one. +func (c *Controller) reportSpeculation( + ctx context.Context, + batch entity.Batch, + set entity.SpeculationPathSet, + snap snapshot, + before entity.SpeculationPathEntry, + hadPassed bool, +) error { + after, hasPassed := livePassedPath(set, snap) + + status, path := entity.RequestStatusSpeculated, after + switch { + case hasPassed: + case hadPassed: + status, path = entity.RequestStatusSpeculating, before + default: + return nil + } + + if err := corerequest.PublishBatchLogs(ctx, c.registry, batch.Queue, batch.Contains, + status, path.ID, map[string]string{ + "batch_id": batch.ID, + "path_id": path.ID, + }, + ); err != nil { + metrics.NamedCounter(c.metricsScope, opName, "request_log_errors", 1) + // Attributed to this head, not the trigger: the loop walks the whole + // queue, so the batch whose members could not be told is usually not the + // one the message named. + return c.attributed(fmt.Errorf("failed to publish request logs for batch %s: %w", batch.ID, err), + entity.BatchSubject(batch.ID)) + } + return nil +} + // finalizeCancellations marks every path of a cancelling batch stopped and, // once they all have, drives the batch to cancelled. // diff --git a/submitqueue/orchestrator/controller/speculate/outcome.go b/submitqueue/orchestrator/controller/speculate/outcome.go index 0491e226..44f60bbb 100644 --- a/submitqueue/orchestrator/controller/speculate/outcome.go +++ b/submitqueue/orchestrator/controller/speculate/outcome.go @@ -91,6 +91,44 @@ func mergeablePath(set entity.SpeculationPathSet, snap snapshot) (entity.Specula return entity.SpeculationPathEntry{}, false } +// passedEntry returns a path whose build passed, regardless of whether reality +// has since contradicted its guesses. +// +// Unlike livePassedPath this asks nothing of the snapshot, which is what makes +// it useful as a before-and-after pair with it: a head that holds a passed entry +// but no live passed path is one whose waiting room a resolving dependency has +// just taken away, and it is back to building. +func passedEntry(set entity.SpeculationPathSet) (entity.SpeculationPathEntry, bool) { + for _, entry := range set.Paths { + if entry.Status == entity.SpeculationPathStatusPassed { + return entry, true + } + } + return entity.SpeculationPathEntry{}, false +} + +// livePassedPath returns a path whose build passed and whose guesses are still +// consistent with how its dependencies are resolving, whether or not they have +// finished resolving. +// +// It is mergeablePath without the settled requirement, and the difference +// between the two is exactly the head's waiting room: work this head had to do +// is done, and all that is left is other batches finishing. Reported rather +// than acted on — nothing may merge on a path this loose, and decide is +// deliberately not built on it. +func livePassedPath(set entity.SpeculationPathSet, snap snapshot) (entity.SpeculationPathEntry, bool) { + for _, entry := range set.Paths { + if entry.Status != entity.SpeculationPathStatusPassed { + continue + } + if assumptionBroken(entry.Path, snap) { + continue + } + return entry, true + } + return entity.SpeculationPathEntry{}, false +} + // allAssumptionsSettled reports whether every dependency has finished the way // the path assumed: one it assumed would succeed has reached Succeeded, and one // it assumed would fail has finished some other way. diff --git a/submitqueue/orchestrator/controller/speculate/outcome_test.go b/submitqueue/orchestrator/controller/speculate/outcome_test.go index 264edaa9..c21e6e43 100644 --- a/submitqueue/orchestrator/controller/speculate/outcome_test.go +++ b/submitqueue/orchestrator/controller/speculate/outcome_test.go @@ -154,6 +154,65 @@ func TestMergeablePath_ExcludesBrokenPassedPath(t *testing.T) { assert.False(t, ok) } +// livePassedPath is mergeablePath without the settled requirement, and the gap +// between the two is the head's waiting room: its own work is done and all that +// remains is other batches finishing. That window is reported to the members, +// so it has to be recognised while mergeablePath still says no. +func TestLivePassedPath(t *testing.T) { + const ( + succeeds = entity.DependencyAssumptionSucceeds + fails = entity.DependencyAssumptionFails + ) + + tests := []struct { + name string + set entity.SpeculationPathSet + dep1State entity.BatchState + dep2State entity.BatchState + want bool + }{ + { + name: "passed with an unsettled dependency is the waiting window", + set: setOf(passedPath(succeeds, succeeds)), + dep1State: entity.BatchStateSpeculating, + dep2State: entity.BatchStateSucceeded, + want: true, + }, + { + name: "passed and fully settled is still passed", + set: setOf(passedPath(succeeds, succeeds)), + dep1State: entity.BatchStateSucceeded, + dep2State: entity.BatchStateSucceeded, + want: true, + }, + { + name: "a build still running is not passed", + set: setOf(entryFor( + pathOver(succeeds, succeeds), entity.SpeculationPathStatusBuilding)), + dep1State: entity.BatchStateSpeculating, + dep2State: entity.BatchStateSucceeded, + want: false, + }, + { + // The head is back to building, which is why losing this is worth + // reporting: the members would otherwise read "speculated" through + // the whole rebuild. + name: "a contradicted assumption takes the path back out", + set: setOf(passedPath(fails, fails)), + dep1State: entity.BatchStateSucceeded, + dep2State: entity.BatchStateSpeculating, + want: false, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + _, ok := livePassedPath(tt.set, snapWith(tt.dep1State, tt.dep2State)) + assert.Equal(t, tt.want, ok) + }) + } +} + func TestHasNoViableFuture(t *testing.T) { headBatch := entity.Batch{ID: head, Dependencies: []string{dep1, dep2}} // With every dependency resolved exactly one assumption pair is unbroken, diff --git a/submitqueue/orchestrator/controller/speculate/run_test.go b/submitqueue/orchestrator/controller/speculate/run_test.go index 172f2788..596a51b5 100644 --- a/submitqueue/orchestrator/controller/speculate/run_test.go +++ b/submitqueue/orchestrator/controller/speculate/run_test.go @@ -84,6 +84,9 @@ type runHarness struct { // messages are the published messages themselves, so a test can assert // what a publish carried and how it was partitioned. messages []entityqueue.Message + // logs holds the request-log entries reported to batch members, kept apart + // from published and messages so pipeline assertions stay about the pipeline. + logs []entity.RequestLog // failTopic, when set, makes every publish to that topic fail. failTopic string } @@ -166,6 +169,12 @@ func newRunHarness(t *testing.T, ctrl *gomock.Controller, spec *scriptedSpeculat if topic == h.failTopic { return assert.AnError } + if topic == "log" { + entry, err := entity.RequestLogFromBytes(msg.Payload) + require.NoError(t, err) + h.logs = append(h.logs, entry) + return nil + } h.published = append(h.published, topic) h.messages = append(h.messages, msg) return nil @@ -179,6 +188,7 @@ func newRunHarness(t *testing.T, ctrl *gomock.Controller, spec *scriptedSpeculat {Key: topickey.TopicKeyMerge, Name: "submitqueue-merge", Queue: q}, {Key: topickey.TopicKeyConclude, Name: "conclude", Queue: q}, {Key: topickey.TopicKeySpeculate, Name: "speculate", Queue: q}, + {Key: topickey.TopicKeyLog, Name: "log", Queue: q}, }) require.NoError(t, err) @@ -1266,3 +1276,74 @@ func TestRun_CancellingPathWithALiveBuildStaysCancelling(t *testing.T) { require.NoError(t, h.run(head)) assert.Empty(t, h.published) } + +// memberHead is a speculating head that actually contains a request, so the +// run's request-log fan-out has somebody to report to. +func memberHead() entity.Batch { + h := speculatingHead() + h.Contains = []string{"q/1"} + return h +} + +// A head whose build passed but whose dependencies have not all settled is in +// the one part of speculation worth naming: its own work is done, and what +// remains is other batches finishing. Without this its members read as still +// building for the whole of that wait. +func TestRun_ReportsPassedPathWhileWaiting(t *testing.T) { + ctrl := gomock.NewController(t) + passed := pathOver(entity.DependencyAssumptionSucceeds, entity.DependencyAssumptionSucceeds) + spec := &scriptedSpeculator{} + + h := newRunHarness(t, ctrl, spec, []entity.Batch{memberHead()}) + h.noBuildsDispatched() + // dep1 has landed the way the path assumed; dep2 has not answered yet, so + // the head cannot merge but has nothing of its own left to run. + h.batches.EXPECT().Get(gomock.Any(), dep1).Return(entity.Batch{ID: dep1, State: entity.BatchStateSucceeded}, nil) + h.batches.EXPECT().Get(gomock.Any(), dep2).Return(entity.Batch{ID: dep2, State: entity.BatchStateSpeculating}, nil) + + entry := entryFor(passed, entity.SpeculationPathStatusPassed) + h.pathSets.EXPECT().Get(gomock.Any(), head).Return(entity.SpeculationPathSet{ + Head: head, + Paths: []entity.SpeculationPathEntry{entry}, + Version: 2, + }, nil).AnyTimes() + + require.NoError(t, h.run(head)) + assert.Empty(t, h.published, "an unsettled head merges nowhere") + + require.Len(t, h.logs, 1) + assert.Equal(t, "q/1", h.logs[0].RequestID) + assert.Equal(t, entity.RequestStatusSpeculated, h.logs[0].Status) + assert.Equal(t, head, h.logs[0].Metadata["batch_id"]) + assert.Equal(t, entry.ID, h.logs[0].Metadata["path_id"]) +} + +// The other half: a dependency that resolves against a passed path's guess +// takes the head's waiting room away and puts it back to building. Reporting +// that is what stops the members reading as speculated through the rebuild. +func TestRun_ReportsBackToSpeculatingWhenPassedPathBreaks(t *testing.T) { + ctrl := gomock.NewController(t) + passed := pathOver(entity.DependencyAssumptionSucceeds, entity.DependencyAssumptionFails) + spec := &scriptedSpeculator{} + + h := newRunHarness(t, ctrl, spec, []entity.Batch{memberHead()}) + h.noBuildsDispatched() + // The path was built without dep2, but dep2 landed after all, so the build + // verified a world that did not happen. + h.batches.EXPECT().Get(gomock.Any(), dep1).Return(entity.Batch{ID: dep1, State: entity.BatchStateSucceeded}, nil) + h.batches.EXPECT().Get(gomock.Any(), dep2).Return(entity.Batch{ID: dep2, State: entity.BatchStateSucceeded}, nil) + + entry := entryFor(passed, entity.SpeculationPathStatusPassed) + h.pathSets.EXPECT().Get(gomock.Any(), head).Return(entity.SpeculationPathSet{ + Head: head, + Paths: []entity.SpeculationPathEntry{entry}, + Version: 2, + }, nil).AnyTimes() + h.pathSets.EXPECT().Update(gomock.Any(), gomock.Any(), int32(2), int32(3)).Return(nil).AnyTimes() + + require.NoError(t, h.run(head)) + + require.Len(t, h.logs, 1) + assert.Equal(t, entity.RequestStatusSpeculating, h.logs[0].Status) + assert.Equal(t, entry.ID, h.logs[0].Metadata["path_id"]) +} diff --git a/submitqueue/orchestrator/controller/speculate/speculate.go b/submitqueue/orchestrator/controller/speculate/speculate.go index 95540770..ee819a4d 100644 --- a/submitqueue/orchestrator/controller/speculate/speculate.go +++ b/submitqueue/orchestrator/controller/speculate/speculate.go @@ -25,6 +25,7 @@ import ( "github.com/uber/submitqueue/platform/metrics" corebatch "github.com/uber/submitqueue/submitqueue/core/batch" "github.com/uber/submitqueue/submitqueue/core/publish" + corerequest "github.com/uber/submitqueue/submitqueue/core/request" "github.com/uber/submitqueue/submitqueue/core/topickey" "github.com/uber/submitqueue/submitqueue/entity" "github.com/uber/submitqueue/submitqueue/extension/speculation/speculator" @@ -144,7 +145,23 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er // the new state and version so the caller keeps writing against a current // copy. Which paths to build for the new head is not decided here — that is // the run's call, taken over the whole queue rather than one batch at a time. +// +// The members are told before the write, not after. This branch runs only from +// Created, so a redelivery of a message whose transition already committed +// skips admit entirely and would never re-publish a log lost on the way out; +// publishing first means a failure nacks with nothing yet changed, and a crash +// in between re-publishes under the same occurrence and dedupes. The cost is +// that a transition which then loses its compare-and-swap leaves one entry for +// a batch that never speculated, superseded by whatever the winner wrote. func (c *Controller) admit(ctx context.Context, store storage.Storage, batch entity.Batch) (entity.Batch, error) { + if err := corerequest.PublishBatchLogs(ctx, c.registry, batch.Queue, batch.Contains, + entity.RequestStatusSpeculating, batch.ID, map[string]string{"batch_id": batch.ID}, + ); err != nil { + metrics.NamedCounter(c.metricsScope, opName, "request_log_errors", 1) + return batch, c.attributed(fmt.Errorf("failed to publish request logs for batch %s: %w", batch.ID, err), + entity.BatchSubject(batch.ID)) + } + // Through Transition for the same reason as applyOutcome: the membership // record has to leave the created bucket with the batch. updated, err := corebatch.Transition(ctx, store, batch, entity.BatchStateSpeculating) diff --git a/submitqueue/orchestrator/controller/speculate/speculate_test.go b/submitqueue/orchestrator/controller/speculate/speculate_test.go index 1d46b0ae..3d8a1bc3 100644 --- a/submitqueue/orchestrator/controller/speculate/speculate_test.go +++ b/submitqueue/orchestrator/controller/speculate/speculate_test.go @@ -16,6 +16,7 @@ package speculate import ( "context" + "fmt" "testing" "github.com/stretchr/testify/assert" @@ -92,6 +93,7 @@ func testBatch(state entity.BatchState, deps ...string) entity.Batch { return entity.Batch{ ID: "test-queue/batch/1", Queue: "test-queue", + Contains: []string{"test-queue/1"}, Dependencies: deps, State: state, Version: 1, @@ -108,6 +110,9 @@ type procHarness struct { builds *storagemock.MockBuildStore spec *quietSpeculator published []string + // logs holds the request-log entries this run reported to batch members, + // kept apart from published so pipeline assertions stay about the pipeline. + logs []entity.RequestLog } func newProcHarness(t *testing.T, ctrl *gomock.Controller, publishErr error) *procHarness { @@ -131,10 +136,16 @@ func newProcHarness(t *testing.T, ctrl *gomock.Controller, publishErr error) *pr pub := queuemock.NewMockPublisher(ctrl) pub.EXPECT().Publish(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn( - func(_ context.Context, topic string, _ entityqueue.Message) error { + func(_ context.Context, topic string, msg entityqueue.Message) error { if publishErr != nil { return publishErr } + if topic == "log" { + entry, err := entity.RequestLogFromBytes(msg.Payload) + require.NoError(t, err) + h.logs = append(h.logs, entry) + return nil + } h.published = append(h.published, topic) return nil }, @@ -147,6 +158,7 @@ func newProcHarness(t *testing.T, ctrl *gomock.Controller, publishErr error) *pr {Key: topickey.TopicKeyMerge, Name: "submitqueue-merge", Queue: q}, {Key: topickey.TopicKeyConclude, Name: "conclude", Queue: q}, {Key: topickey.TopicKeySpeculate, Name: "speculate", Queue: q}, + {Key: topickey.TopicKeyLog, Name: "log", Queue: q}, }) require.NoError(t, err) @@ -193,6 +205,27 @@ func TestProcess_AdmitsCreatedBatch(t *testing.T) { require.NoError(t, h.process(t, ctrl, batch.ID)) assert.Empty(t, h.published, "a batch cannot merge on the message that admitted it") + + // Admission is the first thing a member hears after being batched: without + // it the request reads "batched" for the whole of speculation. + require.Len(t, h.logs, 1) + assert.Equal(t, "test-queue/1", h.logs[0].RequestID) + assert.Equal(t, entity.RequestStatusSpeculating, h.logs[0].Status) + assert.Equal(t, batch.ID, h.logs[0].Metadata["batch_id"]) +} + +// A log that cannot be published must stop admission. This branch runs only +// from Created, so a transition written before a failed publish would never be +// re-reported: the redelivery reads Speculating and skips admit entirely. +func TestProcess_AdmitReportsBeforeTransition(t *testing.T) { + ctrl := gomock.NewController(t) + h := newProcHarness(t, ctrl, fmt.Errorf("enqueue failed")) + batch := testBatch(entity.BatchStateCreated) + + // No Update expectation: the transition must not be reached. + h.batches.EXPECT().Get(gomock.Any(), batch.ID).Return(batch, nil) + + require.Error(t, h.process(t, ctrl, batch.ID)) } // A terminal batch re-publishes to conclude so a lost publish is repaired, and diff --git a/submitqueue/orchestrator/controller/start/start.go b/submitqueue/orchestrator/controller/start/start.go index 0d0349dd..14555db6 100644 --- a/submitqueue/orchestrator/controller/start/start.go +++ b/submitqueue/orchestrator/controller/start/start.go @@ -118,8 +118,8 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er } // Record the "new" status in the request log. - logEntry := entity.NewRequestLog(request.Queue, request.ID, entity.RequestStatusStarted, request.Version, "", nil) - if err := corerequest.PublishLog(ctx, c.registry, logEntry, request.ID); err != nil { + logEntry := entity.NewRequestStatusLog(request.Queue, request.ID, entity.RequestStatusStarted, request.Version, "", nil) + if err := corerequest.PublishLog(ctx, c.registry, logEntry, request.ID, ""); err != nil { metrics.NamedCounter(c.metricsScope, opName, "request_log_errors", 1) return fmt.Errorf("failed to publish request log: %w", err) } diff --git a/submitqueue/orchestrator/controller/validate/validate.go b/submitqueue/orchestrator/controller/validate/validate.go index e5362d32..3e3d7ce2 100644 --- a/submitqueue/orchestrator/controller/validate/validate.go +++ b/submitqueue/orchestrator/controller/validate/validate.go @@ -141,6 +141,16 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er return nil } + // Report that validation has begun. This stage is not instantaneous — the + // merge-conflict check below is an async round trip to runway — so without + // this the request reads "started" for the whole of it. No occurrence: a + // request is validated once, and a redelivery is a retry of that one event. + logEntry := entity.NewRequestStatusLog(request.Queue, request.ID, entity.RequestStatusValidating, 0, "", nil) + if err := corerequest.PublishLog(ctx, c.registry, logEntry, request.ID, ""); err != nil { + coremetrics.NamedCounter(c.metricsScope, "process", "request_log_errors", 1) + return fmt.Errorf("failed to publish request log for %s: %w", request.ID, err) + } + // Duplicate detection: look for any other in-flight request that has already // claimed an overlapping URI in this queue. Per-queue partition leasing // (see platform/consumer + platform/extension/messagequeue) guarantees serial processing within diff --git a/submitqueue/orchestrator/controller/validate/validate_test.go b/submitqueue/orchestrator/controller/validate/validate_test.go index bff40c53..4778d408 100644 --- a/submitqueue/orchestrator/controller/validate/validate_test.go +++ b/submitqueue/orchestrator/controller/validate/validate_test.go @@ -204,15 +204,21 @@ func TestController_Process_PublishesCheckToRunway(t *testing.T) { mockPub := queuemock.NewMockPublisher(ctrl) mockPub.EXPECT().Publish(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn( func(ctx context.Context, topic string, msg entityqueue.Message) error { + if topic != "merge-conflict-check" { + return nil + } gotTopic = topic gotPayload = msg.Payload return nil }, - ) + ).AnyTimes() mockQ := queuemock.NewMockQueue(ctrl) mockQ.EXPECT().Publisher().Return(mockPub).AnyTimes() registry, err := consumer.NewTopicRegistry( - []consumer.TopicConfig{{Key: runwaymq.TopicKeyMergeConflictCheck, Name: "merge-conflict-check", Queue: mockQ}}, + []consumer.TopicConfig{ + {Key: runwaymq.TopicKeyMergeConflictCheck, Name: "merge-conflict-check", Queue: mockQ}, + {Key: topickey.TopicKeyLog, Name: "log", Queue: mockQ}, + }, ) require.NoError(t, err) cpFactory := changeprovidermock.NewMockFactory(ctrl) @@ -606,11 +612,14 @@ func TestController_Process_CustomValidatorPasses(t *testing.T) { logger := zaptest.NewLogger(t).Sugar() mockPub := queuemock.NewMockPublisher(ctrl) - mockPub.EXPECT().Publish(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil) + mockPub.EXPECT().Publish(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).AnyTimes() mockQ := queuemock.NewMockQueue(ctrl) mockQ.EXPECT().Publisher().Return(mockPub).AnyTimes() registry, err := consumer.NewTopicRegistry( - []consumer.TopicConfig{{Key: runwaymq.TopicKeyMergeConflictCheck, Name: "merge-conflict-check", Queue: mockQ}}, + []consumer.TopicConfig{ + {Key: runwaymq.TopicKeyMergeConflictCheck, Name: "merge-conflict-check", Queue: mockQ}, + {Key: topickey.TopicKeyLog, Name: "log", Queue: mockQ}, + }, ) require.NoError(t, err) @@ -715,7 +724,18 @@ func TestController_Process_CustomValidatorFailure_TerminationPublishFails(t *te mockReqStore.EXPECT().Update(gomock.Any(), requestWithState(request, entity.RequestStateError), int32(1), int32(2)).Return(nil) mockPub := queuemock.NewMockPublisher(ctrl) - mockPub.EXPECT().Publish(gomock.Any(), gomock.Any(), gomock.Any()).Return(fmt.Errorf("publish boom")).AnyTimes() + // Fail only the terminal log. The "validating" entry this stage now emits on + // the way in goes to the same topic, so failing every publish would abort + // before validation ever ran and this test would stop covering termination. + mockPub.EXPECT().Publish(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn( + func(_ context.Context, _ string, msg entityqueue.Message) error { + entry, err := entity.RequestLogFromBytes(msg.Payload) + if err == nil && entry.Status == entity.RequestStatusError { + return fmt.Errorf("publish boom") + } + return nil + }, + ).AnyTimes() mockQ := queuemock.NewMockQueue(ctrl) mockQ.EXPECT().Publisher().Return(mockPub).AnyTimes() registry, err := consumer.NewTopicRegistry( diff --git a/test/e2e/submitqueue/harness_test.go b/test/e2e/submitqueue/harness_test.go index 41830fa7..d6e3a19b 100644 --- a/test/e2e/submitqueue/harness_test.go +++ b/test/e2e/submitqueue/harness_test.go @@ -86,16 +86,23 @@ func (s *E2EIntegrationSuite) currentStatus(req request) (entity.RequestStatus, } // awaitStatus polls GetRequestSummaryByID until the request reaches exactly want. -func (s *E2EIntegrationSuite) awaitStatus(req request, want entity.RequestStatus) { +// It returns every distinct status it observed on the way, so a caller can check +// what the summary did and did not report while the request was in flight. +func (s *E2EIntegrationSuite) awaitStatus(req request, want entity.RequestStatus) []entity.RequestStatus { + var seen []entity.RequestStatus pollUntil(persistPollInterval, func() bool { got, err := s.currentStatus(req) if err != nil { s.log.Logf("GetRequestSummaryByID(%s) not ready yet: %v", req.sqid, err) return false } + if len(seen) == 0 || seen[len(seen)-1] != got { + seen = append(seen, got) + } s.log.Logf("GetRequestSummaryByID(%s) = %q (want %q)", req.sqid, got, want) return got == want }) + return seen } // awaitTerminal polls GetRequestSummaryByID until the request reaches a terminal status @@ -116,18 +123,37 @@ func (s *E2EIntegrationSuite) awaitTerminal(req request) entity.RequestStatus { } // timeline returns the ordered customer-facing status history through -// GetRequestHistoryByID. +// GetRequestHistoryByID. Only status entries are included: build events share +// the stream but are not positions, and eventTimeline reads those. func (s *E2EIntegrationSuite) timeline(req request) []entity.RequestStatus { t := s.T() resp, err := s.gatewayClient.GetRequestHistoryByID(s.ctx, &gatewaypb.GetRequestHistoryByIDRequest{Sqid: req.sqid, Queue: req.queue}) require.NoError(t, err, "GetRequestHistoryByID failed for %s", req.sqid) - statuses := make([]entity.RequestStatus, len(resp.Events)) - for i, event := range resp.Events { - statuses[i] = entity.RequestStatus(event.Status) + var statuses []entity.RequestStatus + for _, event := range resp.Events { + if entity.RequestLogType(event.Type) != entity.RequestLogTypeStatus { + continue + } + statuses = append(statuses, entity.RequestStatus(event.Status)) } return statuses } +// eventTimeline returns the ordered build events recorded against the request. +func (s *E2EIntegrationSuite) eventTimeline(req request) []entity.RequestEvent { + t := s.T() + resp, err := s.gatewayClient.GetRequestHistoryByID(s.ctx, &gatewaypb.GetRequestHistoryByIDRequest{Sqid: req.sqid, Queue: req.queue}) + require.NoError(t, err, "GetRequestHistoryByID failed for %s", req.sqid) + var events []entity.RequestEvent + for _, event := range resp.Events { + if entity.RequestLogType(event.Type) != entity.RequestLogTypeEvent { + continue + } + events = append(events, entity.RequestEvent(event.Event)) + } + return events +} + // assertStatusesInOrder asserts that want appears as an ordered subsequence of // the GetRequestHistoryByID status timeline. It tolerates intermediate statuses (so it is // not a change-detector), asserting only the relative order of the statuses that diff --git a/test/e2e/submitqueue/suite_test.go b/test/e2e/submitqueue/suite_test.go index b2d57a17..def81116 100644 --- a/test/e2e/submitqueue/suite_test.go +++ b/test/e2e/submitqueue/suite_test.go @@ -231,7 +231,7 @@ func (s *E2EIntegrationSuite) TestLand_HappyPath_ReachesLanded() { s.log.Logf("Land (happy path) succeeded: sqid=%s; waiting for landed", req.sqid) // Black-box: the customer-facing status reaches landed. - s.awaitStatus(req, entity.RequestStatusLanded) + seen := s.awaitStatus(req, entity.RequestStatusLanded) // Black-box history: all status entries for a request share its request_id // partition on the log topic, and the terminal "landed" is published last. @@ -241,10 +241,34 @@ func (s *E2EIntegrationSuite) TestLand_HappyPath_ReachesLanded() { s.assertStatusesInOrder(req, entity.RequestStatusAccepted, entity.RequestStatusStarted, + entity.RequestStatusValidating, entity.RequestStatusBatched, + entity.RequestStatusSpeculating, + entity.RequestStatusSpeculated, + entity.RequestStatusLanding, entity.RequestStatusLanded, ) + // The two tiers, observed from outside. Build progress shares the stream but + // is not in the status trail above, and the summary never sat on it: a + // request with several paths building at once is speculating, not built. + assert.Subset(s.T(), s.eventTimeline(req), + []entity.RequestEvent{entity.RequestEventBuilding, entity.RequestEventBuilt}, + "GetRequestHistoryByID for %s should record the build that verified it", req.sqid) + + // The types make this unrepresentable in code, so the check is against the + // raw strings and covers the rest of the path — serialize, persist, project, + // read back — where a type cannot follow. Sampling a poll cannot prove a + // value never appeared, so it corroborates the materializer's unit test + // rather than replacing it. + for _, status := range seen { + assert.NotContainsf(s.T(), + []entity.RequestStatus{entity.RequestStatus(entity.RequestEventBuilding), entity.RequestStatus(entity.RequestEventBuilt)}, + status, + "GetRequestSummaryByID for %s reported build progress %q as its current status; observed %v", + req.sqid, status, seen) + } + // White-box (internal state): the operating store's authoritative // RequestState settled on landed. RequestState is point-in-time, so this is a // terminal check, not a sequence. diff --git a/test/integration/submitqueue/gateway/suite_test.go b/test/integration/submitqueue/gateway/suite_test.go index 7b0599df..6e3da761 100644 --- a/test/integration/submitqueue/gateway/suite_test.go +++ b/test/integration/submitqueue/gateway/suite_test.go @@ -292,8 +292,8 @@ func (s *GatewayIntegrationSuite) TestRequestLogConsumer() { Status: entity.RequestStatusAccepting, StatusTimestampMs: 1, Version: 1, Metadata: map[string]string{}, } require.NoError(t, logQueueStore.GetRequestSummaryStore().Create(s.ctx, summary)) - logEntry := entity.NewRequestLog(logQueue, sqid, entity.RequestStatusStarted, 1, "", nil) - require.NoError(t, corerequest.PublishLog(s.ctx, registry, logEntry, sqid), + logEntry := entity.NewRequestStatusLog(logQueue, sqid, entity.RequestStatusStarted, 1, "", nil) + require.NoError(t, corerequest.PublishLog(s.ctx, registry, logEntry, sqid, ""), "failed to publish request log to log topic") s.log.Logf("Published 'started' log for sqid=%s; waiting for gateway consumer to persist it", sqid)