| 1 | package jobs |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "fmt" |
| 6 | "io" |
| 7 | "reasonix/internal/event" |
| 8 | "reasonix/internal/nilutil" |
| 9 | "strings" |
| 10 | ) |
| 11 | |
| 12 | // StartForSession launches a job owned by parentSession. Session-scoped readers |
| 13 | // only see jobs whose owner matches the active session. |
| 14 | func (m *Manager) StartForSession(parentSession, kind, label string, run func(ctx context.Context, out io.Writer) (string, error)) *Job { |
| 15 | parentSession = strings.TrimSpace(parentSession) |
| 16 | kind = strings.TrimSpace(kind) |
| 17 | if err := validatePathSegment(parentSession, "parentSession"); err != nil { |
| 18 | return m.startInvalid(parentSession, kind, label, err) |
| 19 | } |
| 20 | if err := validatePathSegment(kind, "kind"); err != nil { |
| 21 | return m.startInvalid(parentSession, kind, label, err) |
| 22 | } |
| 23 | m.mu.Lock() |
| 24 | m.seq++ |
| 25 | id := fmt.Sprintf("%s-%d", kind, m.seq) |
| 26 | ctx, cancel := context.WithCancel(m.root) |
| 27 | startedAt := nowMs() |
| 28 | logPath, metaPath, file, artifactErr := m.openArtifactLocked(parentSession, id) |
| 29 | j := &Job{ |
| 30 | ID: id, |
| 31 | Kind: kind, |
| 32 | Label: label, |
| 33 | SessionID: parentSession, |
| 34 | status: Running, |
| 35 | clock: jobClock{startedAt: startedAt, activityAt: startedAt}, |
| 36 | cancel: cancel, |
| 37 | done: make(chan struct{}), |
| 38 | artifactPath: logPath, |
| 39 | artifactMetaPath: metaPath, |
| 40 | artifactFile: file, |
| 41 | artifactComplete: artifactErr == "", |
| 42 | artifactErr: artifactErr, |
| 43 | } |
| 44 | ctx = WithSession(ctx, parentSession) |
| 45 | ctx = context.WithValue(ctx, jobCtxKey{}, j) |
| 46 | key := jobKey(parentSession, id) |
| 47 | m.jobs[key] = j |
| 48 | m.order = append(m.order, key) |
| 49 | m.mu.Unlock() |
| 50 | j.mu.Lock() |
| 51 | if err := m.writeJobMetaLocked(j, Running); err != nil { |
| 52 | j.artifactComplete = false |
| 53 | j.artifactErr = err.Error() |
| 54 | } |
| 55 | j.mu.Unlock() |
| 56 | if m.onJobStart != nil { |
| 57 | m.onJobStart(j.done) |
| 58 | } |
| 59 | |
| 60 | m.emitIfActive(parentSession, event.Event{Kind: event.Notice, Level: event.LevelInfo, Text: startedText(kind, id, label)}) |
| 61 | m.notifyRuntime(parentSession, id) |
| 62 | |
| 63 | if !nilutil.IsNil(m.taskRecorder) { |
| 64 | m.taskRecorder.RecordStart(id, kind, label) |
| 65 | } |
| 66 | |
| 67 | m.wg.Add(1) |
| 68 | if m.stalledWarning > 0 { |
| 69 | m.wg.Add(1) |
| 70 | go m.monitorStalled(parentSession, j) |
| 71 | } |
| 72 | go m.runJob(ctx, j, run) |
| 73 | return j |
| 74 | } |
| 75 | |
| 76 | func (m *Manager) runJob(ctx context.Context, j *Job, run func(context.Context, io.Writer) (string, error)) { |
| 77 | defer m.wg.Done() |
| 78 | result, err := runRecovered(ctx, jobWriter{j}, run) |
| 79 | j.mu.Lock() |
| 80 | j.outcome.returned = true |
| 81 | j.mu.Unlock() |
| 82 | |
| 83 | var st Status |
| 84 | switch { |
| 85 | case ctx.Err() != nil: |
| 86 | st = Killed |
| 87 | case err != nil: |
| 88 | st = Failed |
| 89 | if result == "" { |
| 90 | result = err.Error() |
| 91 | } |
| 92 | default: |
| 93 | st = Done |
| 94 | } |
| 95 | finishedAt := nowMs() |
| 96 | if result != "" { |
| 97 | j.mu.Lock() |
| 98 | if j.artifactFile != nil { |
| 99 | if _, writeErr := j.artifactFile.WriteString(result); writeErr != nil { |
| 100 | j.artifactErr = writeErr.Error() |
| 101 | } |
| 102 | } else { |
| 103 | j.outcome.text = result |
| 104 | } |
| 105 | j.tail = appendTail(j.tail, []byte(result), defaultTailBytes) |
| 106 | j.mu.Unlock() |
| 107 | } |
| 108 | targetDir := m.artifactTargetDirForJob(j) |
| 109 | j.mu.Lock() |
| 110 | if j.artifactFile != nil { |
| 111 | if closeErr := j.artifactFile.Close(); closeErr != nil && j.artifactErr == "" { |
| 112 | j.artifactErr = closeErr.Error() |
| 113 | } |
| 114 | j.artifactFile = nil |
| 115 | } |
| 116 | if j.artifactErr != "" { |
| 117 | j.artifactComplete = false |
| 118 | } |
| 119 | j.clock.finishedAt = finishedAt |
| 120 | if targetDir != "" { |
| 121 | if moveErr := j.moveArtifactToDirLocked(targetDir); moveErr != nil { |
| 122 | j.noteArtifactErr("migration: " + moveErr.Error()) |
| 123 | } |
| 124 | } |
| 125 | metaErr := m.writeJobMetaLocked(j, st) |
| 126 | if metaErr != nil { |
| 127 | j.noteArtifactErr("metadata: " + metaErr.Error()) |
| 128 | } |
| 129 | j.mu.Unlock() |
| 130 | // Queue the drain note and closing Notice before terminal status so Wait |
| 131 | // cannot observe completion before DrainCompletedNote sees its bookkeeping. |
| 132 | // The structured runtime notification follows the actual done boundary. |
| 133 | parentSession := m.recordCompletion(j, st, err) |
| 134 | |
| 135 | j.mu.Lock() |
| 136 | if j.status != Killed { // a concurrent Kill already published Killed — keep it |
| 137 | j.status = st |
| 138 | } |
| 139 | if j.artifactPath != "" && j.artifactComplete { |
| 140 | j.outcome.text = "" |
| 141 | j.tail = nil |
| 142 | } |
| 143 | j.mu.Unlock() |
| 144 | close(j.done) |
| 145 | m.notifyRuntime(parentSession, j.ID) |
| 146 | } |
| 147 |