diff --git a/spindle/db/events.go b/spindle/db/events.go index d6bb93467..dc03dd212 100644 --- a/spindle/db/events.go +++ b/spindle/db/events.go @@ -87,7 +87,7 @@ func (d *DB) CompleteMillLease( n *notifier.Notifier, ) error { return d.ApplyEventBatch(n, func(tx *EventBatchTx) error { - if err := tx.InsertStatusEvent(workflowId, status, workflowError, exitCode); err != nil { + if err := tx.InsertStatusChange(workflowId, status, workflowError, exitCode); err != nil { return err } return tx.DeleteLease(leaseID) diff --git a/spindle/db/mill_state.go b/spindle/db/mill_state.go index 6b79fda62..fe599af0f 100644 --- a/spindle/db/mill_state.go +++ b/spindle/db/mill_state.go @@ -3,6 +3,7 @@ package db import ( "context" "database/sql" + "errors" "fmt" "time" @@ -280,6 +281,23 @@ func (tx *EventBatchTx) InsertStatusEvent(wid models.WorkflowId, status string, return insertStatusTx(tx.tx, wid, models.StatusKind(status), workflowError, exitCode) } +func (tx *EventBatchTx) InsertStatusChange(wid models.WorkflowId, status string, workflowError *string, exitCode *int64) error { + var latest string + err := tx.tx.QueryRow( + `select status from workflow_statuses + where pipeline_id = ? and workflow = ? + order by id desc limit 1`, + string(wid.PipelineId), wid.Name, + ).Scan(&latest) + if err != nil && !errors.Is(err, sql.ErrNoRows) { + return err + } + if latest == status { + return nil + } + return tx.InsertStatusEvent(wid, status, workflowError, exitCode) +} + func (tx *EventBatchTx) DeleteLease(leaseID string) error { _, err := tx.tx.Exec(`delete from mill_leases where lease_id = ?`, leaseID) return err diff --git a/spindle/mill/executor/executor.go b/spindle/mill/executor/executor.go index 5d75265c3..c9f63fe0d 100644 --- a/spindle/mill/executor/executor.go +++ b/spindle/mill/executor/executor.go @@ -87,6 +87,7 @@ type Executor struct { lifecycleCtx context.Context jobsWG sync.WaitGroup + finishWG sync.WaitGroup } type reservation struct { @@ -109,6 +110,8 @@ type reservation struct { failureClass string failureReason string millRecordsTerminalMetrics bool + millAcceptsOutcome bool + reportedStatus string } type messageEncoder interface { @@ -209,6 +212,7 @@ reconnect: stopObserver() <-observerDone e.drainEvents(&cursor) + e.finishWG.Wait() } func (e *Executor) runSession(ctx context.Context) error { @@ -632,6 +636,7 @@ func (e *Executor) handleCommit(ctx context.Context, cl *millv1.CommitLease) { res.wf.CacheBindings = bindings res.committed = true res.millRecordsTerminalMetrics = cl.GetMillRecordsTerminalMetrics() + res.millAcceptsOutcome = cl.GetAcceptsAttemptOutcome() if res.ttlTimer != nil { res.ttlTimer.Stop() } diff --git a/spindle/mill/executor/observe.go b/spindle/mill/executor/observe.go index 0cee58173..a4256d327 100644 --- a/spindle/mill/executor/observe.go +++ b/spindle/mill/executor/observe.go @@ -57,12 +57,54 @@ func (e *Executor) onStatusRow(st *db.StatusRow) error { } if models.StatusKind(st.Status).IsFinish() { - return e.finishJob(res, st) + reported, err := e.reportOutcome(res, st) + if err != nil { + e.l.Error("report outcome failed", "lease", res.leaseID, "err", err) + } + if !reported { + return e.finishJob(res, st) + } + // a debug shell can hold the VM for minutes, and waiting here stalls every other job's status + e.finishWG.Add(1) + go func() { + defer e.finishWG.Done() + if err := e.finishJob(res, st); err != nil { + e.l.Error("finish job failed", "lease", res.leaseID, "err", err) + } + }() + return nil } return e.appendStatus(res.leaseID, st) } +// false when the run already ended or the mill predates outcomes +func (e *Executor) reportOutcome(res *reservation, st *db.StatusRow) (bool, error) { + e.mu.Lock() + if e.active[res.leaseID] != res || !res.millAcceptsOutcome || res.runDone == nil { + e.mu.Unlock() + return false, nil + } + select { + case <-res.runDone: + e.mu.Unlock() + return false, nil + default: + } + status := st.Status + if res.cancelled { + status = string(models.StatusKindCancelled) + } + res.reportedStatus = status + e.mu.Unlock() + + // every log line has to reach the mill before the status does + if res.stopTail != nil { + res.stopTail() + } + return true, e.appendOutcome(res.leaseID, status, st) +} + func (e *Executor) finishJob(res *reservation, st *db.StatusRow) error { e.mu.Lock() if e.active[res.leaseID] != res { @@ -70,6 +112,7 @@ func (e *Executor) finishJob(res *reservation, st *db.StatusRow) error { return nil } cancelled := res.cancelled + reported := res.reportedStatus runDone := res.runDone e.mu.Unlock() @@ -91,7 +134,11 @@ func (e *Executor) finishJob(res *reservation, st *db.StatusRow) error { if cancelled { terminalStatus = string(models.StatusKindCancelled) } - if cancelled { + if reported != "" { + // the mill already published it, so a late cancel can't change it + terminalStatus = reported + } + if cancelled && terminalStatus == string(models.StatusKindCancelled) { failureClass = string(engine.FailureClassUser) failureReason = string(engine.FailureReasonCancelled) } diff --git a/spindle/mill/executor/outbox.go b/spindle/mill/executor/outbox.go index 24e38828e..2c6e06ea4 100644 --- a/spindle/mill/executor/outbox.go +++ b/spindle/mill/executor/outbox.go @@ -74,6 +74,8 @@ func (e *Executor) appendAndSend(leaseID string, payload any, control bool) erro entry.Payload = payload case *millv1.Event_AttemptResult: entry.Payload = payload + case *millv1.Event_AttemptOutcome: + entry.Payload = payload case *millv1.Event_CacheUpdate: entry.Payload = payload default: @@ -85,7 +87,7 @@ func (e *Executor) appendAndSend(leaseID string, payload any, control bool) erro case *millv1.Event_AttemptResult: durable = true isTerminal = true - case *millv1.Event_CacheUpdate: + case *millv1.Event_AttemptOutcome, *millv1.Event_CacheUpdate: durable = true } entry.Seqno = ^uint64(0) @@ -186,24 +188,44 @@ func defaultTerminalAttribution(status string) (string, string) { } } -func (e *Executor) appendTerminalWithArtifact( - leaseID, status string, - st *db.StatusRow, - ref, hash, failureClass, failureReason string, - millRecordsTerminalMetrics bool, -) error { - var terminalStatus millv1.TerminalStatus +func terminalStatusProto(status string) (millv1.TerminalStatus, error) { switch status { case string(models.StatusKindSuccess): - terminalStatus = millv1.TerminalStatus_TERMINAL_STATUS_SUCCESS + return millv1.TerminalStatus_TERMINAL_STATUS_SUCCESS, nil case string(models.StatusKindFailed): - terminalStatus = millv1.TerminalStatus_TERMINAL_STATUS_FAILED + return millv1.TerminalStatus_TERMINAL_STATUS_FAILED, nil case string(models.StatusKindTimeout): - terminalStatus = millv1.TerminalStatus_TERMINAL_STATUS_TIMEOUT + return millv1.TerminalStatus_TERMINAL_STATUS_TIMEOUT, nil case string(models.StatusKindCancelled): - terminalStatus = millv1.TerminalStatus_TERMINAL_STATUS_CANCELLED + return millv1.TerminalStatus_TERMINAL_STATUS_CANCELLED, nil default: - return fmt.Errorf("unsupported terminal status %q", status) + return 0, fmt.Errorf("unsupported terminal status %q", status) + } +} + +func (e *Executor) appendOutcome(leaseID, status string, st *db.StatusRow) error { + terminalStatus, err := terminalStatusProto(status) + if err != nil { + return err + } + exit, errStr := parseStatusExitAndError(st) + payload := &millv1.Event_AttemptOutcome{AttemptOutcome: &millv1.AttemptOutcome{ + Status: terminalStatus, + Error: errStr, + ExitCode: exit, + }} + return e.appendAndSend(leaseID, payload, true) +} + +func (e *Executor) appendTerminalWithArtifact( + leaseID, status string, + st *db.StatusRow, + ref, hash, failureClass, failureReason string, + millRecordsTerminalMetrics bool, +) error { + terminalStatus, err := terminalStatusProto(status) + if err != nil { + return err } exit, errStr := parseStatusExitAndError(st) diff --git a/spindle/mill/executor/reserved_test.go b/spindle/mill/executor/reserved_test.go index f51367485..fdf16fc70 100644 --- a/spindle/mill/executor/reserved_test.go +++ b/spindle/mill/executor/reserved_test.go @@ -1032,6 +1032,98 @@ func TestFinishJobWaitsForEngineCleanup(t *testing.T) { } } +func outboxEvents(t *testing.T, d *db.DB) []*millv1.Event { + t.Helper() + rows, err := d.ListOutboxRows() + if err != nil { + t.Fatal(err) + } + events := make([]*millv1.Event, 0, len(rows)) + for _, row := range rows { + var entry millv1.Event + if err := proto.Unmarshal(row.Payload, &entry); err != nil { + t.Fatal(err) + } + events = append(events, &entry) + } + return events +} + +func TestFinishedStatusGoesOutBeforeEngineCleanup(t *testing.T) { + e := testExecutor(t) + res := &reservation{ + leaseID: "lease-1", + wid: models.WorkflowId{PipelineId: models.PipelineId("r"), Name: "build"}, + runDone: make(chan struct{}), + millAcceptsOutcome: true, + } + e.active[res.leaseID] = res + + returned := make(chan error, 1) + go func() { + returned <- e.onStatusRow(&db.StatusRow{ + Pipeline: string(res.wid.PipelineId), + Workflow: res.wid.Name, + Status: string(models.StatusKindFailed), + }) + }() + select { + case err := <-returned: + if err != nil { + t.Fatal(err) + } + case <-time.After(time.Second): + t.Fatal("the status loop waited on engine cleanup") + } + + events := outboxEvents(t, e.db) + if len(events) != 1 || events[0].GetAttemptOutcome().GetStatus() != millv1.TerminalStatus_TERMINAL_STATUS_FAILED { + t.Fatalf("events before cleanup = %v, want one failed outcome", events) + } + + e.mu.Lock() + res.cancelled = true + e.mu.Unlock() + close(res.runDone) + e.finishWG.Wait() + + events = outboxEvents(t, e.db) + if len(events) != 2 || events[1].GetAttemptResult().GetStatus() != millv1.TerminalStatus_TERMINAL_STATUS_FAILED { + t.Fatalf("events after cleanup = %v, want the outcome then a failed result", events) + } +} + +func TestOlderMillGetsNoOutcome(t *testing.T) { + e := testExecutor(t) + res := &reservation{ + leaseID: "lease-1", + wid: models.WorkflowId{PipelineId: models.PipelineId("r"), Name: "build"}, + runDone: make(chan struct{}), + } + e.active[res.leaseID] = res + + returned := make(chan error, 1) + go func() { + returned <- e.onStatusRow(&db.StatusRow{ + Pipeline: string(res.wid.PipelineId), + Workflow: res.wid.Name, + Status: string(models.StatusKindFailed), + }) + }() + time.Sleep(50 * time.Millisecond) + if events := outboxEvents(t, e.db); len(events) != 0 { + t.Fatalf("events before cleanup = %v, an older mill rejects outcomes", events) + } + close(res.runDone) + if err := <-returned; err != nil { + t.Fatal(err) + } + events := outboxEvents(t, e.db) + if len(events) != 1 || events[0].GetAttemptResult() == nil { + t.Fatalf("events = %v, want only the result", events) + } +} + func TestReplayRejectsMalformedOutboxRow(t *testing.T) { d := testDB(t) e := &Executor{ diff --git a/spindle/mill/mill.go b/spindle/mill/mill.go index 176161cb7..7d0d02436 100644 --- a/spindle/mill/mill.go +++ b/spindle/mill/mill.go @@ -1013,6 +1013,7 @@ func (m *Mill) commitAndWait(ctx context.Context, wf *models.Workflow, unlocked Tracestate: tracestate, MillRecordsTerminalMetrics: lease.millRecordsTerminalMetrics, CacheBindings: cacheBindingsToProto(wf.CacheBindings), + AcceptsAttemptOutcome: true, }} // commit retries ride reconnects, a reservation outlives one @@ -1107,6 +1108,33 @@ func (m *Mill) waitCommitRetry(ctx context.Context, lease *RemoteLease) (bool, e } } +func terminalStatusString(status millv1.TerminalStatus) (string, error) { + switch status { + case millv1.TerminalStatus_TERMINAL_STATUS_SUCCESS: + return string(models.StatusKindSuccess), nil + case millv1.TerminalStatus_TERMINAL_STATUS_FAILED: + return string(models.StatusKindFailed), nil + case millv1.TerminalStatus_TERMINAL_STATUS_TIMEOUT: + return string(models.StatusKindTimeout), nil + case millv1.TerminalStatus_TERMINAL_STATUS_CANCELLED: + return string(models.StatusKindCancelled), nil + default: + return "", protoErrf("unsupported terminal status %v", status) + } +} + +func statusDetails(errMsg string, exitCode int64) (*string, *int64) { + var e *string + if errMsg != "" { + e = &errMsg + } + var c *int64 + if exitCode != 0 { + c = &exitCode + } + return e, c +} + func terminalError(status millv1.TerminalStatus) error { switch status { case millv1.TerminalStatus_TERMINAL_STATUS_SUCCESS: @@ -1632,32 +1660,30 @@ func (m *Mill) onEventBatch(sess *millSession, batch *millv1.EventBatch) error { } } - case entry.GetAttemptResult() != nil: - ar := entry.GetAttemptResult() - statusStr := "success" - switch ar.Status { - case millv1.TerminalStatus_TERMINAL_STATUS_SUCCESS: - statusStr = "success" - case millv1.TerminalStatus_TERMINAL_STATUS_FAILED: - statusStr = "failed" - case millv1.TerminalStatus_TERMINAL_STATUS_TIMEOUT: - statusStr = "timeout" - case millv1.TerminalStatus_TERMINAL_STATUS_CANCELLED: - statusStr = "cancelled" - default: - return batchApply{}, protoErrf("unsupported terminal status %v", ar.Status) + case entry.GetAttemptOutcome() != nil: + out := entry.GetAttemptOutcome() + statusStr, err := terminalStatusString(out.GetStatus()) + if err != nil { + return batchApply{}, err } - var errMsg *string - if e := ar.GetError(); e != "" { - errMsg = &e + if tx != nil { + errMsg, exitCode := statusDetails(out.GetError(), out.GetExitCode()) + if err := tx.InsertStatusChange(lease.wid, statusStr, errMsg, exitCode); err != nil { + return batchApply{}, err + } } - var exitCode *int64 - if c := ar.GetExitCode(); c != 0 { - exitCode = &c + + case entry.GetAttemptResult() != nil: + ar := entry.GetAttemptResult() + statusStr, err := terminalStatusString(ar.GetStatus()) + if err != nil { + return batchApply{}, err } finishedInBatch[lease.id] = struct{}{} if tx != nil { - if err := tx.InsertStatusEvent(lease.wid, statusStr, errMsg, exitCode); err != nil { + // an AttemptOutcome may have published this status already + errMsg, exitCode := statusDetails(ar.GetError(), ar.GetExitCode()) + if err := tx.InsertStatusChange(lease.wid, statusStr, errMsg, exitCode); err != nil { return batchApply{}, err } if err := tx.DeleteLease(lease.id); err != nil { diff --git a/spindle/mill/mill_test.go b/spindle/mill/mill_test.go index 2cb6aeb0b..c90d2a568 100644 --- a/spindle/mill/mill_test.go +++ b/spindle/mill/mill_test.go @@ -753,6 +753,69 @@ func TestAtomicBatchRollback(t *testing.T) { } } +func TestAttemptOutcomePublishesStatusBeforeTheLeaseEnds(t *testing.T) { + m, bdb := restoreTestMill(t, Config{ReconnectGrace: time.Minute}) + sess := newSession("node-1", "inc-1", nil, nopEncoder(), discardLogger()) + m.attachSession(sess) + + owned := newLease("lease-1", "node-1", "inc-1", "dummy") + owned.wid = models.WorkflowId{PipelineId: models.PipelineId("r"), Name: "build"} + m.mu.Lock() + m.leases[owned.id] = owned + m.mu.Unlock() + + if err := m.onEventBatch(sess, &millv1.EventBatch{ + Epoch: sess.epoch, + Events: []*millv1.Event{{ + Seqno: 1, + LeaseId: owned.id, + Payload: &millv1.Event_AttemptOutcome{AttemptOutcome: &millv1.AttemptOutcome{ + Status: millv1.TerminalStatus_TERMINAL_STATUS_FAILED, + Error: "User step error: exited with code 1", + ExitCode: 1, + }}, + }}, + }); err != nil { + t.Fatal(err) + } + if st, err := bdb.GetStatus(owned.wid); err != nil || st != models.StatusKindFailed { + t.Fatalf("status after outcome = %q (err %v), want failed", st, err) + } + if _, ok := pollTerminal(owned); ok { + t.Fatal("an outcome finished the lease, the VM is still held") + } + if owned.getState() == leaseDone { + t.Fatal("an outcome sealed the lease") + } + + if err := m.onEventBatch(sess, &millv1.EventBatch{ + Epoch: sess.epoch, + Events: []*millv1.Event{{ + Seqno: 2, + LeaseId: owned.id, + Payload: &millv1.Event_AttemptResult{AttemptResult: &millv1.AttemptResult{ + Status: millv1.TerminalStatus_TERMINAL_STATUS_FAILED, + Error: "User step error: exited with code 1", + }}, + }}, + }); err != nil { + t.Fatal(err) + } + if _, ok := pollTerminal(owned); !ok { + t.Fatal("the result after an outcome was not delivered") + } + var rows int + if err := bdb.QueryRow( + `select count(*) from workflow_statuses where pipeline_id = ? and workflow = ? and status = 'failed'`, + string(owned.wid.PipelineId), owned.wid.Name, + ).Scan(&rows); err != nil { + t.Fatal(err) + } + if rows != 1 { + t.Fatalf("failed status rows = %d, want 1", rows) + } +} + func TestCacheUpdatesCommitWithStreamCursor(t *testing.T) { m, bdb := restoreTestMill(t, Config{ReconnectGrace: time.Minute}) cache, err := storage.NewDisk(t.TempDir()) diff --git a/spindle/mill/proto/gen/mill.pb.go b/spindle/mill/proto/gen/mill.pb.go index 2b035a406..4372c2608 100644 --- a/spindle/mill/proto/gen/mill.pb.go +++ b/spindle/mill/proto/gen/mill.pb.go @@ -927,8 +927,11 @@ type CommitLease struct { Tracestate string `protobuf:"bytes,4,opt,name=tracestate,proto3" json:"tracestate,omitempty"` MillRecordsTerminalMetrics bool `protobuf:"varint,5,opt,name=mill_records_terminal_metrics,json=millRecordsTerminalMetrics,proto3" json:"mill_records_terminal_metrics,omitempty"` CacheBindings []*CacheBinding `protobuf:"bytes,6,rep,name=cache_bindings,json=cacheBindings,proto3" json:"cache_bindings,omitempty"` - unknownFields protoimpl.UnknownFields - sizeCache protoimpl.SizeCache + // older mills drop the session on an unknown event payload, so executors only + // send AttemptOutcome when this is set + AcceptsAttemptOutcome bool `protobuf:"varint,7,opt,name=accepts_attempt_outcome,json=acceptsAttemptOutcome,proto3" json:"accepts_attempt_outcome,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache } func (x *CommitLease) Reset() { @@ -1003,6 +1006,13 @@ func (x *CommitLease) GetCacheBindings() []*CacheBinding { return nil } +func (x *CommitLease) GetAcceptsAttemptOutcome() bool { + if x != nil { + return x.AcceptsAttemptOutcome + } + return false +} + type Committed struct { state protoimpl.MessageState `protogen:"open.v1"` LeaseId string `protobuf:"bytes,1,opt,name=lease_id,json=leaseId,proto3" json:"lease_id,omitempty"` @@ -1394,6 +1404,68 @@ func (x *AttemptResult) GetMillRecordsTerminalMetrics() bool { return false } +// the result is final but the executor still holds the VM, so the lease stays +// charged until the AttemptResult +type AttemptOutcome struct { + state protoimpl.MessageState `protogen:"open.v1"` + Status TerminalStatus `protobuf:"varint,1,opt,name=status,proto3,enum=spindle.mill.v1.TerminalStatus" json:"status,omitempty"` + Error string `protobuf:"bytes,2,opt,name=error,proto3" json:"error,omitempty"` + ExitCode int64 `protobuf:"varint,3,opt,name=exit_code,json=exitCode,proto3" json:"exit_code,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *AttemptOutcome) Reset() { + *x = AttemptOutcome{} + mi := &file_spindle_mill_v1_mill_proto_msgTypes[16] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *AttemptOutcome) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*AttemptOutcome) ProtoMessage() {} + +func (x *AttemptOutcome) ProtoReflect() protoreflect.Message { + mi := &file_spindle_mill_v1_mill_proto_msgTypes[16] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use AttemptOutcome.ProtoReflect.Descriptor instead. +func (*AttemptOutcome) Descriptor() ([]byte, []int) { + return file_spindle_mill_v1_mill_proto_rawDescGZIP(), []int{16} +} + +func (x *AttemptOutcome) GetStatus() TerminalStatus { + if x != nil { + return x.Status + } + return TerminalStatus_TERMINAL_STATUS_UNSPECIFIED +} + +func (x *AttemptOutcome) GetError() string { + if x != nil { + return x.Error + } + return "" +} + +func (x *AttemptOutcome) GetExitCode() int64 { + if x != nil { + return x.ExitCode + } + return 0 +} + type CacheUpdate struct { state protoimpl.MessageState `protogen:"open.v1"` Action CacheUpdateAction `protobuf:"varint,1,opt,name=action,proto3,enum=spindle.mill.v1.CacheUpdateAction" json:"action,omitempty"` @@ -1406,7 +1478,7 @@ type CacheUpdate struct { func (x *CacheUpdate) Reset() { *x = CacheUpdate{} - mi := &file_spindle_mill_v1_mill_proto_msgTypes[16] + mi := &file_spindle_mill_v1_mill_proto_msgTypes[17] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1418,7 +1490,7 @@ func (x *CacheUpdate) String() string { func (*CacheUpdate) ProtoMessage() {} func (x *CacheUpdate) ProtoReflect() protoreflect.Message { - mi := &file_spindle_mill_v1_mill_proto_msgTypes[16] + mi := &file_spindle_mill_v1_mill_proto_msgTypes[17] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1431,7 +1503,7 @@ func (x *CacheUpdate) ProtoReflect() protoreflect.Message { // Deprecated: Use CacheUpdate.ProtoReflect.Descriptor instead. func (*CacheUpdate) Descriptor() ([]byte, []int) { - return file_spindle_mill_v1_mill_proto_rawDescGZIP(), []int{16} + return file_spindle_mill_v1_mill_proto_rawDescGZIP(), []int{17} } func (x *CacheUpdate) GetAction() CacheUpdateAction { @@ -1473,7 +1545,7 @@ type LiveLog struct { func (x *LiveLog) Reset() { *x = LiveLog{} - mi := &file_spindle_mill_v1_mill_proto_msgTypes[17] + mi := &file_spindle_mill_v1_mill_proto_msgTypes[18] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1485,7 +1557,7 @@ func (x *LiveLog) String() string { func (*LiveLog) ProtoMessage() {} func (x *LiveLog) ProtoReflect() protoreflect.Message { - mi := &file_spindle_mill_v1_mill_proto_msgTypes[17] + mi := &file_spindle_mill_v1_mill_proto_msgTypes[18] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1498,7 +1570,7 @@ func (x *LiveLog) ProtoReflect() protoreflect.Message { // Deprecated: Use LiveLog.ProtoReflect.Descriptor instead. func (*LiveLog) Descriptor() ([]byte, []int) { - return file_spindle_mill_v1_mill_proto_rawDescGZIP(), []int{17} + return file_spindle_mill_v1_mill_proto_rawDescGZIP(), []int{18} } func (x *LiveLog) GetLeaseId() string { @@ -1525,6 +1597,7 @@ type Event struct { // *Event_StatusEvent // *Event_AttemptResult // *Event_CacheUpdate + // *Event_AttemptOutcome Payload isEvent_Payload `protobuf_oneof:"payload"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache @@ -1532,7 +1605,7 @@ type Event struct { func (x *Event) Reset() { *x = Event{} - mi := &file_spindle_mill_v1_mill_proto_msgTypes[18] + mi := &file_spindle_mill_v1_mill_proto_msgTypes[19] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1544,7 +1617,7 @@ func (x *Event) String() string { func (*Event) ProtoMessage() {} func (x *Event) ProtoReflect() protoreflect.Message { - mi := &file_spindle_mill_v1_mill_proto_msgTypes[18] + mi := &file_spindle_mill_v1_mill_proto_msgTypes[19] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1557,7 +1630,7 @@ func (x *Event) ProtoReflect() protoreflect.Message { // Deprecated: Use Event.ProtoReflect.Descriptor instead. func (*Event) Descriptor() ([]byte, []int) { - return file_spindle_mill_v1_mill_proto_rawDescGZIP(), []int{18} + return file_spindle_mill_v1_mill_proto_rawDescGZIP(), []int{19} } func (x *Event) GetSeqno() uint64 { @@ -1608,6 +1681,15 @@ func (x *Event) GetCacheUpdate() *CacheUpdate { return nil } +func (x *Event) GetAttemptOutcome() *AttemptOutcome { + if x != nil { + if x, ok := x.Payload.(*Event_AttemptOutcome); ok { + return x.AttemptOutcome + } + } + return nil +} + type isEvent_Payload interface { isEvent_Payload() } @@ -1624,12 +1706,18 @@ type Event_CacheUpdate struct { CacheUpdate *CacheUpdate `protobuf:"bytes,5,opt,name=cache_update,json=cacheUpdate,proto3,oneof"` } +type Event_AttemptOutcome struct { + AttemptOutcome *AttemptOutcome `protobuf:"bytes,6,opt,name=attempt_outcome,json=attemptOutcome,proto3,oneof"` +} + func (*Event_StatusEvent) isEvent_Payload() {} func (*Event_AttemptResult) isEvent_Payload() {} func (*Event_CacheUpdate) isEvent_Payload() {} +func (*Event_AttemptOutcome) isEvent_Payload() {} + // a flushed bundle of events type EventBatch struct { state protoimpl.MessageState `protogen:"open.v1"` @@ -1641,7 +1729,7 @@ type EventBatch struct { func (x *EventBatch) Reset() { *x = EventBatch{} - mi := &file_spindle_mill_v1_mill_proto_msgTypes[19] + mi := &file_spindle_mill_v1_mill_proto_msgTypes[20] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1653,7 +1741,7 @@ func (x *EventBatch) String() string { func (*EventBatch) ProtoMessage() {} func (x *EventBatch) ProtoReflect() protoreflect.Message { - mi := &file_spindle_mill_v1_mill_proto_msgTypes[19] + mi := &file_spindle_mill_v1_mill_proto_msgTypes[20] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1666,7 +1754,7 @@ func (x *EventBatch) ProtoReflect() protoreflect.Message { // Deprecated: Use EventBatch.ProtoReflect.Descriptor instead. func (*EventBatch) Descriptor() ([]byte, []int) { - return file_spindle_mill_v1_mill_proto_rawDescGZIP(), []int{19} + return file_spindle_mill_v1_mill_proto_rawDescGZIP(), []int{20} } func (x *EventBatch) GetEpoch() string { @@ -1694,7 +1782,7 @@ type Ack struct { func (x *Ack) Reset() { *x = Ack{} - mi := &file_spindle_mill_v1_mill_proto_msgTypes[20] + mi := &file_spindle_mill_v1_mill_proto_msgTypes[21] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1706,7 +1794,7 @@ func (x *Ack) String() string { func (*Ack) ProtoMessage() {} func (x *Ack) ProtoReflect() protoreflect.Message { - mi := &file_spindle_mill_v1_mill_proto_msgTypes[20] + mi := &file_spindle_mill_v1_mill_proto_msgTypes[21] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1719,7 +1807,7 @@ func (x *Ack) ProtoReflect() protoreflect.Message { // Deprecated: Use Ack.ProtoReflect.Descriptor instead. func (*Ack) Descriptor() ([]byte, []int) { - return file_spindle_mill_v1_mill_proto_rawDescGZIP(), []int{20} + return file_spindle_mill_v1_mill_proto_rawDescGZIP(), []int{21} } func (x *Ack) GetEpoch() string { @@ -1759,7 +1847,7 @@ type Message struct { func (x *Message) Reset() { *x = Message{} - mi := &file_spindle_mill_v1_mill_proto_msgTypes[21] + mi := &file_spindle_mill_v1_mill_proto_msgTypes[22] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1771,7 +1859,7 @@ func (x *Message) String() string { func (*Message) ProtoMessage() {} func (x *Message) ProtoReflect() protoreflect.Message { - mi := &file_spindle_mill_v1_mill_proto_msgTypes[21] + mi := &file_spindle_mill_v1_mill_proto_msgTypes[22] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1784,7 +1872,7 @@ func (x *Message) ProtoReflect() protoreflect.Message { // Deprecated: Use Message.ProtoReflect.Descriptor instead. func (*Message) Descriptor() ([]byte, []int) { - return file_spindle_mill_v1_mill_proto_rawDescGZIP(), []int{21} + return file_spindle_mill_v1_mill_proto_rawDescGZIP(), []int{22} } func (x *Message) GetHello() *Hello { @@ -1907,7 +1995,7 @@ type QuotaRequest struct { func (x *QuotaRequest) Reset() { *x = QuotaRequest{} - mi := &file_spindle_mill_v1_mill_proto_msgTypes[22] + mi := &file_spindle_mill_v1_mill_proto_msgTypes[23] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1919,7 +2007,7 @@ func (x *QuotaRequest) String() string { func (*QuotaRequest) ProtoMessage() {} func (x *QuotaRequest) ProtoReflect() protoreflect.Message { - mi := &file_spindle_mill_v1_mill_proto_msgTypes[22] + mi := &file_spindle_mill_v1_mill_proto_msgTypes[23] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1932,7 +2020,7 @@ func (x *QuotaRequest) ProtoReflect() protoreflect.Message { // Deprecated: Use QuotaRequest.ProtoReflect.Descriptor instead. func (*QuotaRequest) Descriptor() ([]byte, []int) { - return file_spindle_mill_v1_mill_proto_rawDescGZIP(), []int{22} + return file_spindle_mill_v1_mill_proto_rawDescGZIP(), []int{23} } func (x *QuotaRequest) GetRequestId() string { @@ -1999,7 +2087,7 @@ type QuotaResponse struct { func (x *QuotaResponse) Reset() { *x = QuotaResponse{} - mi := &file_spindle_mill_v1_mill_proto_msgTypes[23] + mi := &file_spindle_mill_v1_mill_proto_msgTypes[24] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -2011,7 +2099,7 @@ func (x *QuotaResponse) String() string { func (*QuotaResponse) ProtoMessage() {} func (x *QuotaResponse) ProtoReflect() protoreflect.Message { - mi := &file_spindle_mill_v1_mill_proto_msgTypes[23] + mi := &file_spindle_mill_v1_mill_proto_msgTypes[24] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -2024,7 +2112,7 @@ func (x *QuotaResponse) ProtoReflect() protoreflect.Message { // Deprecated: Use QuotaResponse.ProtoReflect.Descriptor instead. func (*QuotaResponse) Descriptor() ([]byte, []int) { - return file_spindle_mill_v1_mill_proto_rawDescGZIP(), []int{23} + return file_spindle_mill_v1_mill_proto_rawDescGZIP(), []int{24} } func (x *QuotaResponse) GetRequestId() string { @@ -2148,7 +2236,7 @@ const file_spindle_mill_v1_mill_proto_rawDesc = "" + "\asave_id\x18\x06 \x01(\tB\a\xbaH\x04r\x02\x10\x01R\x06saveId\x12\"\n" + "\bsave_key\x18\a \x01(\tB\a\xbaH\x04r\x02\x10\x01R\asaveKey\x12)\n" + "\x10restore_checksum\x18\b \x01(\tR\x0frestoreChecksum\x12,\n" + - "\x12restore_size_bytes\x18\t \x01(\x03R\x10restoreSizeBytes\"\xaf\x02\n" + + "\x12restore_size_bytes\x18\t \x01(\x03R\x10restoreSizeBytes\"\xe7\x02\n" + "\vCommitLease\x12\"\n" + "\blease_id\x18\x01 \x01(\tB\a\xbaH\x04r\x02\x10\x01R\aleaseId\x121\n" + "\asecrets\x18\x02 \x03(\v2\x17.spindle.mill.v1.SecretR\asecrets\x12 \n" + @@ -2157,7 +2245,8 @@ const file_spindle_mill_v1_mill_proto_rawDesc = "" + "tracestate\x18\x04 \x01(\tR\n" + "tracestate\x12A\n" + "\x1dmill_records_terminal_metrics\x18\x05 \x01(\bR\x1amillRecordsTerminalMetrics\x12D\n" + - "\x0ecache_bindings\x18\x06 \x03(\v2\x1d.spindle.mill.v1.CacheBindingR\rcacheBindings\"/\n" + + "\x0ecache_bindings\x18\x06 \x03(\v2\x1d.spindle.mill.v1.CacheBindingR\rcacheBindings\x126\n" + + "\x17accepts_attempt_outcome\x18\a \x01(\bR\x15acceptsAttemptOutcome\"/\n" + "\tCommitted\x12\"\n" + "\blease_id\x18\x01 \x01(\tB\a\xbaH\x04r\x02\x10\x01R\aleaseId\"2\n" + "\fReleaseLease\x12\"\n" + @@ -2183,7 +2272,12 @@ const file_spindle_mill_v1_mill_proto_rawDesc = "" + "\flog_artifact\x18\x04 \x01(\v2\x1c.spindle.mill.v1.LogArtifactR\vlogArtifact\x12,\n" + "\rfailure_class\x18\x05 \x01(\tB\a\xbaH\x04r\x02( R\ffailureClass\x12.\n" + "\x0efailure_reason\x18\x06 \x01(\tB\a\xbaH\x04r\x02(@R\rfailureReason\x12A\n" + - "\x1dmill_records_terminal_metrics\x18\a \x01(\bR\x1amillRecordsTerminalMetrics\"\xb2\x01\n" + + "\x1dmill_records_terminal_metrics\x18\a \x01(\bR\x1amillRecordsTerminalMetrics\"\x93\x01\n" + + "\x0eAttemptOutcome\x12C\n" + + "\x06status\x18\x01 \x01(\x0e2\x1f.spindle.mill.v1.TerminalStatusB\n" + + "\xbaH\a\x82\x01\x04\x10\x01 \x00R\x06status\x12\x1f\n" + + "\x05error\x18\x02 \x01(\tB\t\xbaH\x06r\x04(\x80\x80\x04R\x05error\x12\x1b\n" + + "\texit_code\x18\x03 \x01(\x03R\bexitCode\"\xb2\x01\n" + "\vCacheUpdate\x12F\n" + "\x06action\x18\x01 \x01(\x0e2\".spindle.mill.v1.CacheUpdateActionB\n" + "\xbaH\a\x82\x01\x04\x10\x01 \x00R\x06action\x12\x17\n" + @@ -2193,13 +2287,14 @@ const file_spindle_mill_v1_mill_proto_rawDesc = "" + "\bchecksum\x18\x05 \x01(\tR\bchecksum\"Q\n" + "\aLiveLog\x12\"\n" + "\blease_id\x18\x01 \x01(\tB\a\xbaH\x04r\x02\x10\x01R\aleaseId\x12\"\n" + - "\braw_json\x18\x02 \x01(\fB\a\xbaH\x04z\x02\x10\x01R\arawJson\"\xab\x02\n" + + "\braw_json\x18\x02 \x01(\fB\a\xbaH\x04z\x02\x10\x01R\arawJson\"\xf7\x02\n" + "\x05Event\x12\x1d\n" + "\x05seqno\x18\x01 \x01(\x04B\a\xbaH\x042\x02 \x00R\x05seqno\x12\"\n" + "\blease_id\x18\x02 \x01(\tB\a\xbaH\x04r\x02\x10\x01R\aleaseId\x12A\n" + "\fstatus_event\x18\x03 \x01(\v2\x1c.spindle.mill.v1.StatusEventH\x00R\vstatusEvent\x12G\n" + "\x0eattempt_result\x18\x04 \x01(\v2\x1e.spindle.mill.v1.AttemptResultH\x00R\rattemptResult\x12A\n" + - "\fcache_update\x18\x05 \x01(\v2\x1c.spindle.mill.v1.CacheUpdateH\x00R\vcacheUpdateB\x10\n" + + "\fcache_update\x18\x05 \x01(\v2\x1c.spindle.mill.v1.CacheUpdateH\x00R\vcacheUpdate\x12J\n" + + "\x0fattempt_outcome\x18\x06 \x01(\v2\x1f.spindle.mill.v1.AttemptOutcomeH\x00R\x0eattemptOutcomeB\x10\n" + "\apayload\x12\x05\xbaH\x02\b\x01\"e\n" + "\n" + "EventBatch\x12\x1d\n" + @@ -2309,7 +2404,7 @@ func file_spindle_mill_v1_mill_proto_rawDescGZIP() []byte { } var file_spindle_mill_v1_mill_proto_enumTypes = make([]protoimpl.EnumInfo, 5) -var file_spindle_mill_v1_mill_proto_msgTypes = make([]protoimpl.MessageInfo, 28) +var file_spindle_mill_v1_mill_proto_msgTypes = make([]protoimpl.MessageInfo, 29) var file_spindle_mill_v1_mill_proto_goTypes = []any{ (RejectClass)(0), // 0: spindle.mill.v1.RejectClass (NonterminalStatus)(0), // 1: spindle.mill.v1.NonterminalStatus @@ -2332,57 +2427,60 @@ var file_spindle_mill_v1_mill_proto_goTypes = []any{ (*StatusEvent)(nil), // 18: spindle.mill.v1.StatusEvent (*LogArtifact)(nil), // 19: spindle.mill.v1.LogArtifact (*AttemptResult)(nil), // 20: spindle.mill.v1.AttemptResult - (*CacheUpdate)(nil), // 21: spindle.mill.v1.CacheUpdate - (*LiveLog)(nil), // 22: spindle.mill.v1.LiveLog - (*Event)(nil), // 23: spindle.mill.v1.Event - (*EventBatch)(nil), // 24: spindle.mill.v1.EventBatch - (*Ack)(nil), // 25: spindle.mill.v1.Ack - (*Message)(nil), // 26: spindle.mill.v1.Message - (*QuotaRequest)(nil), // 27: spindle.mill.v1.QuotaRequest - (*QuotaResponse)(nil), // 28: spindle.mill.v1.QuotaResponse - nil, // 29: spindle.mill.v1.EngineAvailability.LoadEntry - nil, // 30: spindle.mill.v1.NodeSnapshot.EnginesEntry - nil, // 31: spindle.mill.v1.ReserveResult.QuotaResourcesEntry - nil, // 32: spindle.mill.v1.QuotaRequest.ResourcesEntry + (*AttemptOutcome)(nil), // 21: spindle.mill.v1.AttemptOutcome + (*CacheUpdate)(nil), // 22: spindle.mill.v1.CacheUpdate + (*LiveLog)(nil), // 23: spindle.mill.v1.LiveLog + (*Event)(nil), // 24: spindle.mill.v1.Event + (*EventBatch)(nil), // 25: spindle.mill.v1.EventBatch + (*Ack)(nil), // 26: spindle.mill.v1.Ack + (*Message)(nil), // 27: spindle.mill.v1.Message + (*QuotaRequest)(nil), // 28: spindle.mill.v1.QuotaRequest + (*QuotaResponse)(nil), // 29: spindle.mill.v1.QuotaResponse + nil, // 30: spindle.mill.v1.EngineAvailability.LoadEntry + nil, // 31: spindle.mill.v1.NodeSnapshot.EnginesEntry + nil, // 32: spindle.mill.v1.ReserveResult.QuotaResourcesEntry + nil, // 33: spindle.mill.v1.QuotaRequest.ResourcesEntry } var file_spindle_mill_v1_mill_proto_depIdxs = []int32{ - 29, // 0: spindle.mill.v1.EngineAvailability.load:type_name -> spindle.mill.v1.EngineAvailability.LoadEntry - 30, // 1: spindle.mill.v1.NodeSnapshot.engines:type_name -> spindle.mill.v1.NodeSnapshot.EnginesEntry + 30, // 0: spindle.mill.v1.EngineAvailability.load:type_name -> spindle.mill.v1.EngineAvailability.LoadEntry + 31, // 1: spindle.mill.v1.NodeSnapshot.engines:type_name -> spindle.mill.v1.NodeSnapshot.EnginesEntry 0, // 2: spindle.mill.v1.ReserveResult.reject_class:type_name -> spindle.mill.v1.RejectClass - 31, // 3: spindle.mill.v1.ReserveResult.quota_resources:type_name -> spindle.mill.v1.ReserveResult.QuotaResourcesEntry + 32, // 3: spindle.mill.v1.ReserveResult.quota_resources:type_name -> spindle.mill.v1.ReserveResult.QuotaResourcesEntry 11, // 4: spindle.mill.v1.CommitLease.secrets:type_name -> spindle.mill.v1.Secret 12, // 5: spindle.mill.v1.CommitLease.cache_bindings:type_name -> spindle.mill.v1.CacheBinding 1, // 6: spindle.mill.v1.StatusEvent.status:type_name -> spindle.mill.v1.NonterminalStatus 2, // 7: spindle.mill.v1.AttemptResult.status:type_name -> spindle.mill.v1.TerminalStatus 19, // 8: spindle.mill.v1.AttemptResult.log_artifact:type_name -> spindle.mill.v1.LogArtifact - 3, // 9: spindle.mill.v1.CacheUpdate.action:type_name -> spindle.mill.v1.CacheUpdateAction - 18, // 10: spindle.mill.v1.Event.status_event:type_name -> spindle.mill.v1.StatusEvent - 20, // 11: spindle.mill.v1.Event.attempt_result:type_name -> spindle.mill.v1.AttemptResult - 21, // 12: spindle.mill.v1.Event.cache_update:type_name -> spindle.mill.v1.CacheUpdate - 23, // 13: spindle.mill.v1.EventBatch.events:type_name -> spindle.mill.v1.Event - 5, // 14: spindle.mill.v1.Message.hello:type_name -> spindle.mill.v1.Hello - 6, // 15: spindle.mill.v1.Message.resume:type_name -> spindle.mill.v1.Resume - 8, // 16: spindle.mill.v1.Message.node_snapshot:type_name -> spindle.mill.v1.NodeSnapshot - 9, // 17: spindle.mill.v1.Message.reserve_seat:type_name -> spindle.mill.v1.ReserveSeat - 10, // 18: spindle.mill.v1.Message.reserve_result:type_name -> spindle.mill.v1.ReserveResult - 13, // 19: spindle.mill.v1.Message.commit_lease:type_name -> spindle.mill.v1.CommitLease - 14, // 20: spindle.mill.v1.Message.committed:type_name -> spindle.mill.v1.Committed - 15, // 21: spindle.mill.v1.Message.release_lease:type_name -> spindle.mill.v1.ReleaseLease - 16, // 22: spindle.mill.v1.Message.cancel_attempt:type_name -> spindle.mill.v1.CancelAttempt - 17, // 23: spindle.mill.v1.Message.cancel_ack:type_name -> spindle.mill.v1.CancelAck - 24, // 24: spindle.mill.v1.Message.event_batch:type_name -> spindle.mill.v1.EventBatch - 25, // 25: spindle.mill.v1.Message.ack:type_name -> spindle.mill.v1.Ack - 22, // 26: spindle.mill.v1.Message.live_log:type_name -> spindle.mill.v1.LiveLog - 27, // 27: spindle.mill.v1.Message.quota_req:type_name -> spindle.mill.v1.QuotaRequest - 28, // 28: spindle.mill.v1.Message.quota_resp:type_name -> spindle.mill.v1.QuotaResponse - 4, // 29: spindle.mill.v1.QuotaRequest.operation:type_name -> spindle.mill.v1.QuotaOperation - 32, // 30: spindle.mill.v1.QuotaRequest.resources:type_name -> spindle.mill.v1.QuotaRequest.ResourcesEntry - 7, // 31: spindle.mill.v1.NodeSnapshot.EnginesEntry.value:type_name -> spindle.mill.v1.EngineAvailability - 32, // [32:32] is the sub-list for method output_type - 32, // [32:32] is the sub-list for method input_type - 32, // [32:32] is the sub-list for extension type_name - 32, // [32:32] is the sub-list for extension extendee - 0, // [0:32] is the sub-list for field type_name + 2, // 9: spindle.mill.v1.AttemptOutcome.status:type_name -> spindle.mill.v1.TerminalStatus + 3, // 10: spindle.mill.v1.CacheUpdate.action:type_name -> spindle.mill.v1.CacheUpdateAction + 18, // 11: spindle.mill.v1.Event.status_event:type_name -> spindle.mill.v1.StatusEvent + 20, // 12: spindle.mill.v1.Event.attempt_result:type_name -> spindle.mill.v1.AttemptResult + 22, // 13: spindle.mill.v1.Event.cache_update:type_name -> spindle.mill.v1.CacheUpdate + 21, // 14: spindle.mill.v1.Event.attempt_outcome:type_name -> spindle.mill.v1.AttemptOutcome + 24, // 15: spindle.mill.v1.EventBatch.events:type_name -> spindle.mill.v1.Event + 5, // 16: spindle.mill.v1.Message.hello:type_name -> spindle.mill.v1.Hello + 6, // 17: spindle.mill.v1.Message.resume:type_name -> spindle.mill.v1.Resume + 8, // 18: spindle.mill.v1.Message.node_snapshot:type_name -> spindle.mill.v1.NodeSnapshot + 9, // 19: spindle.mill.v1.Message.reserve_seat:type_name -> spindle.mill.v1.ReserveSeat + 10, // 20: spindle.mill.v1.Message.reserve_result:type_name -> spindle.mill.v1.ReserveResult + 13, // 21: spindle.mill.v1.Message.commit_lease:type_name -> spindle.mill.v1.CommitLease + 14, // 22: spindle.mill.v1.Message.committed:type_name -> spindle.mill.v1.Committed + 15, // 23: spindle.mill.v1.Message.release_lease:type_name -> spindle.mill.v1.ReleaseLease + 16, // 24: spindle.mill.v1.Message.cancel_attempt:type_name -> spindle.mill.v1.CancelAttempt + 17, // 25: spindle.mill.v1.Message.cancel_ack:type_name -> spindle.mill.v1.CancelAck + 25, // 26: spindle.mill.v1.Message.event_batch:type_name -> spindle.mill.v1.EventBatch + 26, // 27: spindle.mill.v1.Message.ack:type_name -> spindle.mill.v1.Ack + 23, // 28: spindle.mill.v1.Message.live_log:type_name -> spindle.mill.v1.LiveLog + 28, // 29: spindle.mill.v1.Message.quota_req:type_name -> spindle.mill.v1.QuotaRequest + 29, // 30: spindle.mill.v1.Message.quota_resp:type_name -> spindle.mill.v1.QuotaResponse + 4, // 31: spindle.mill.v1.QuotaRequest.operation:type_name -> spindle.mill.v1.QuotaOperation + 33, // 32: spindle.mill.v1.QuotaRequest.resources:type_name -> spindle.mill.v1.QuotaRequest.ResourcesEntry + 7, // 33: spindle.mill.v1.NodeSnapshot.EnginesEntry.value:type_name -> spindle.mill.v1.EngineAvailability + 34, // [34:34] is the sub-list for method output_type + 34, // [34:34] is the sub-list for method input_type + 34, // [34:34] is the sub-list for extension type_name + 34, // [34:34] is the sub-list for extension extendee + 0, // [0:34] is the sub-list for field type_name } func init() { file_spindle_mill_v1_mill_proto_init() } @@ -2390,10 +2488,11 @@ func file_spindle_mill_v1_mill_proto_init() { if File_spindle_mill_v1_mill_proto != nil { return } - file_spindle_mill_v1_mill_proto_msgTypes[18].OneofWrappers = []any{ + file_spindle_mill_v1_mill_proto_msgTypes[19].OneofWrappers = []any{ (*Event_StatusEvent)(nil), (*Event_AttemptResult)(nil), (*Event_CacheUpdate)(nil), + (*Event_AttemptOutcome)(nil), } type x struct{} out := protoimpl.TypeBuilder{ @@ -2401,7 +2500,7 @@ func file_spindle_mill_v1_mill_proto_init() { GoPackagePath: reflect.TypeOf(x{}).PkgPath(), RawDescriptor: unsafe.Slice(unsafe.StringData(file_spindle_mill_v1_mill_proto_rawDesc), len(file_spindle_mill_v1_mill_proto_rawDesc)), NumEnums: 5, - NumMessages: 28, + NumMessages: 29, NumExtensions: 0, NumServices: 0, }, diff --git a/spindle/mill/proto/spindle/mill/v1/mill.proto b/spindle/mill/proto/spindle/mill/v1/mill.proto index fafb4ed79..711ae329c 100644 --- a/spindle/mill/proto/spindle/mill/v1/mill.proto +++ b/spindle/mill/proto/spindle/mill/v1/mill.proto @@ -103,6 +103,9 @@ message CommitLease { string tracestate = 4; bool mill_records_terminal_metrics = 5; repeated CacheBinding cache_bindings = 6; + // older mills drop the session on an unknown event payload, so executors only + // send AttemptOutcome when this is set + bool accepts_attempt_outcome = 7; } message Committed { @@ -165,6 +168,17 @@ message AttemptResult { bool mill_records_terminal_metrics = 7; } +// the result is final but the executor still holds the VM, so the lease stays +// charged until the AttemptResult +message AttemptOutcome { + TerminalStatus status = 1 [(buf.validate.field).enum = { + defined_only: true + not_in: 0 + }]; + string error = 2 [(buf.validate.field).string.max_bytes = 65536]; + int64 exit_code = 3; +} + enum CacheUpdateAction { CACHE_UPDATE_ACTION_UNSPECIFIED = 0; CACHE_UPDATE_ACTION_USED = 1; @@ -199,6 +213,7 @@ message Event { StatusEvent status_event = 3; AttemptResult attempt_result = 4; CacheUpdate cache_update = 5; + AttemptOutcome attempt_outcome = 6; } } diff --git a/spindle/wipe.go b/spindle/wipe.go index 009494527..51e7a9d5e 100644 --- a/spindle/wipe.go +++ b/spindle/wipe.go @@ -138,12 +138,13 @@ func (s *Spindle) WipeOwner(ctx context.Context, ownerDid syntax.DID, reason str func (s *Spindle) cancelRepoWorkflows(ctx context.Context, repoDid syntax.DID, reason string) ([]models.WorkflowId, error) { var wids []models.WorkflowId seen := map[models.WorkflowId]bool{} - add := func(rkey, name string) { + add := func(rkey, name string) models.WorkflowId { wid := models.WorkflowId{PipelineId: models.PipelineId(rkey), Name: name} if !seen[wid] { seen[wid] = true wids = append(wids, wid) } + return wid } pipes, err := s.db.ListPipelineWorkflows(repoDid.String()) @@ -158,19 +159,24 @@ func (s *Spindle) cancelRepoWorkflows(ctx context.Context, repoDid syntax.DID, r if err != nil { return nil, fmt.Errorf("list mill leases: %w", err) } + leased := map[models.WorkflowId]bool{} for _, lease := range leases { if lease.RepoDID == repoDid.String() { - add(lease.PipelineID, lease.Workflow) + leased[add(lease.PipelineID, lease.Workflow)] = true } } var errs []error for _, wid := range wids { st, err := s.db.GetStatus(wid) - if err != nil || !st.IsFinish() { + finished := err == nil && st.IsFinish() + if !finished { if err := s.db.StatusCancelled(wid, reason, -1, s.n); err != nil { errs = append(errs, err) } + } + // a finished workflow keeps its lease while a debug shell holds the VM + if !finished || leased[wid] { engine.CancelWorkflow(wid) for _, eng := range s.engs { if err := eng.DestroyWorkflow(ctx, wid); err != nil { diff --git a/spindle/wipe_test.go b/spindle/wipe_test.go index 3eb070c3a..01b9db269 100644 --- a/spindle/wipe_test.go +++ b/spindle/wipe_test.go @@ -314,6 +314,39 @@ func TestWipeOwnerWipesAllOwnedRepos(t *testing.T) { } } +type destroyRecorder struct { + models.Engine + destroyed []models.WorkflowId +} + +func (e *destroyRecorder) DestroyWorkflow(_ context.Context, wid models.WorkflowId) error { + e.destroyed = append(e.destroyed, wid) + return nil +} + +func TestWipeDestroysFinishedWorkflowStillHoldingItsVM(t *testing.T) { + s := newWipeTestSpindle(t) + repoDid := syntax.DID("did:plc:repo1") + seedWipeRepo(t, s, repoDid, "did:plc:owner1") + eng := &destroyRecorder{} + s.engs["microvm"] = eng + + wid := models.WorkflowId{PipelineId: models.PipelineId("p1"), Name: "w1"} + if err := s.db.StatusFailed(wid, "User step error: exited with code 1", 1, s.n); err != nil { + t.Fatal(err) + } + + if _, err := s.cancelRepoWorkflows(context.Background(), repoDid, "banned"); err != nil { + t.Fatal(err) + } + if len(eng.destroyed) != 1 || eng.destroyed[0] != wid { + t.Fatalf("destroyed = %v, want the held workflow", eng.destroyed) + } + if st, err := s.db.GetStatus(wid); err != nil || st != models.StatusKindFailed { + t.Fatalf("status = %q (err %v), want the failure kept", st, err) + } +} + func ptr[T any](v T) *T { return &v } func TestIngestAccountTakedownWipes(t *testing.T) {