Something went wrong. Try again.
Monorepo for Tangled tangled.org
Something went wrong. Try again.
1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253package db
import ( "context" "database/sql" "time"
"tangled.org/core/notifier" "tangled.org/core/spindle/models")
// no row and no lease means a dead process, because workers delete the row once they take a job// run this before the workers start, or live ones look orphanedfunc (d *DB) FailOrphanedWorkflows(n *notifier.Notifier) (int64, error) { res, err := d.Exec(` insert into workflow_statuses (pipeline_id, workflow, status, error, exit_code, created_at) select ws.pipeline_id, ws.workflow, ?, ?, -1, ? from workflow_statuses ws where ws.status in ('pending', 'running') and ws.id = ( select max(latest.id) from workflow_statuses latest where latest.pipeline_id = ws.pipeline_id and latest.workflow = ws.workflow ) and ws.pipeline_id not in (select pipeline_id from jobs) and (ws.pipeline_id, ws.workflow) not in (select pipeline_id, workflow from mill_leases)`, string(models.StatusKindFailed), "spindle restarted before this workflow finished", time.Now().Format(time.RFC3339Nano), ) if err != nil { return 0, err } failed, err := res.RowsAffected() if err == nil && failed > 0 { n.NotifyAll() } return failed, err}
func failPipelineTx(ctx context.Context, tx *sql.Tx, pipelineID, reason string) error { _, err := tx.ExecContext(ctx, ` insert into workflow_statuses (pipeline_id, workflow, status, error, exit_code, created_at) select ws.pipeline_id, ws.workflow, ?, ?, -1, ? from workflow_statuses ws where ws.pipeline_id = ? and ws.status in ('pending', 'running') and ws.id = ( select max(latest.id) from workflow_statuses latest where latest.pipeline_id = ws.pipeline_id and latest.workflow = ws.workflow )`, string(models.StatusKindFailed), reason, time.Now().Format(time.RFC3339Nano), pipelineID, ) return err}