+1

go.mod

reviewed

··· 45 45 github.com/prometheus/procfs v0.19.2 // indirect 46 46 github.com/redis/go-redis/v9 v9.7.3 // indirect 47 47 github.com/rivo/uniseg v0.4.7 // indirect 48 48 + github.com/rogpeppe/go-internal v1.14.1 // indirect 48 49 github.com/spaolacci/murmur3 v1.1.0 // indirect 49 50 github.com/whyrusleeping/cbor-gen v0.3.1 // indirect 50 51 github.com/xo/terminfo v0.0.0-20220910002029-abceb7e1c41e // indirect

+6 -2

go.sum

reviewed

··· 8 8 github.com/bluesky-social/indigo v0.0.0-20260220055544-bf41e2ee75ab/go.mod h1:VG/LeqLGNI3Ew7lsYixajnZGFfWPv144qbUddh+Oyag= 9 9 github.com/bluesky-social/jetstream v0.0.0-20260226214936-e0274250f654 h1:OK76FcHhZp8ohjRB0OMWgti0oYAWFlt3KDQcIkH1pfI= 10 10 github.com/bluesky-social/jetstream v0.0.0-20260226214936-e0274250f654/go.mod h1:vt8kVRKtvrBspt9G38wDD8+BotjIMO8u8IYoVnyE4zY= 11 11 + github.com/bsm/ginkgo/v2 v2.12.0 h1:Ny8MWAHyOepLGlLKYmXG4IEkioBysk6GpaRTLC8zwWs= 12 12 + github.com/bsm/ginkgo/v2 v2.12.0/go.mod h1:SwYbGRRDovPVboqFv0tPTcG1sN61LM1Z4ARdbAV9g4c= 13 13 + github.com/bsm/gomega v1.27.10 h1:yeMWxP2pV2fG3FgAODIY8EiRE3dy0aeFYt4l7wh6yKA= 14 14 + github.com/bsm/gomega v1.27.10/go.mod h1:JyEr/xRbxbtgWNi8tIEVPUYZ5Dzef52k01W3YH0H+O0= 11 15 github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs= 12 16 github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= 13 17 github.com/charmbracelet/colorprofile v0.2.3-0.20250311203215-f60798e515dc h1:4pZI35227imm7yK2bGPcfpFEmuY1gc2YSTShr4iJBfs= ··· 85 89 github.com/rivo/uniseg v0.2.0/go.mod h1:J6wj4VEh+S6ZtnVlnTBMWIodfgj8LQOQFoIToxlJtxc= 86 90 github.com/rivo/uniseg v0.4.7 h1:WUdvkW8uEhrYfLC4ZzdpI2ztxP1I582+49Oc5Mq64VQ= 87 91 github.com/rivo/uniseg v0.4.7/go.mod h1:FN3SvrM+Zdj16jyLfmOkMNblXMcoc8DfTHruCPUcx88= 88 88 - github.com/rogpeppe/go-internal v1.12.0 h1:exVL4IDcn6na9z1rAb56Vxr+CgyK3nn3O+epU5NdKM8= 89 89 - github.com/rogpeppe/go-internal v1.12.0/go.mod h1:E+RYuTGaKKdloAfM02xzb0FW3Paa99yedzYV+kq4uf4= 92 92 + github.com/rogpeppe/go-internal v1.14.1 h1:UQB4HGPB6osV0SQTLymcB4TgvyWu6ZyliaW0tI/otEQ= 93 93 + github.com/rogpeppe/go-internal v1.14.1/go.mod h1:MaRKkUm5W0goXpeCfT7UZI6fk/L7L7so1lCWt35ZSgc= 90 94 github.com/spaolacci/murmur3 v1.1.0 h1:7c1g84S4BPRrfL5Xrdp6fOJ206sU9y293DDHaoy0bLI= 91 95 github.com/spaolacci/murmur3 v1.1.0/go.mod h1:JwIasOWyU6f++ZhiEuf87xNszmSA2myDM2Kzu9HwQUA= 92 96 github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U=

+149 -2

http.go

reviewed

··· 2 2 3 3 // HTTP surface of the spindle. 4 4 // 5 5 - // Three roles to keep in mind: 5 5 + // Four roles to keep in mind: 6 6 // 7 7 // 1. Verification: the Tangled appview hits /xrpc/sh.tangled.owner during 8 8 // spindle registration to confirm the operator owns this instance. ··· 13 13 // 3. Webhooks: Buildkite POSTs build/job state changes to 14 14 // /webhooks/buildkite, which we'll translate into pipeline.status 15 15 // events on (2). 16 16 + // 4. Logs: GET /logs/{knot}/{pipelineRkey}/{workflow} delegates to the 17 17 + // configured Provider so the appview (or a curling operator) can 18 18 + // pull captured workflow output for a specific run. 16 19 17 20 import ( 18 21 "context" ··· 35 38 // The logger is read from ctx via loggerFrom. The broker is the 36 39 // in-process pub/sub used by /events to fan published records out to 37 40 // connected websocket subscribers. 38 38 - func runHTTP(ctx context.Context, cfg config, br *broker) error { 41 41 + func runHTTP(ctx context.Context, cfg config, br *broker, provider Provider) error { 39 42 logger := loggerFrom(ctx) 40 43 41 44 mux := http.NewServeMux() 42 45 mux.HandleFunc("GET /", rootHandler()) 43 46 mux.HandleFunc("GET /events", eventsHandler(logger, br)) 47 47 + mux.HandleFunc("GET /logs/{knot}/{pipelineRkey}/{workflow}", logsHandler(logger, provider)) 44 48 mux.HandleFunc("GET /xrpc/"+tangled.OwnerNSID, ownerHandler(logger, cfg.OwnerDID)) 45 49 mux.HandleFunc("POST /webhooks/buildkite", buildkiteWebhookHandler()) 46 50 ··· 97 101 func buildkiteWebhookHandler() http.HandlerFunc { 98 102 return func(w http.ResponseWriter, r *http.Request) { 99 103 http.Error(w, "not implemented", http.StatusNotImplemented) 104 104 + } 105 105 + } 106 106 + 107 107 + // logsHandler serves captured workflow logs over a WebSocket, 108 108 + // matching the wire protocol of the upstream Tangled spindle so the 109 109 + // appview's log proxy (appview/pipelines.Logs) treats us as a drop-in 110 110 + // source. The path shape is 111 111 + // 112 112 + // GET /logs/{knot}/{pipelineRkey}/{workflow} 113 113 + // 114 114 + // which matches the (knot, pipelineRkey, workflow) tuple 115 115 + // Provider.Spawn is invoked with — the same identity used in the 116 116 + // pipeline ATURI. Workflow names commonly contain a dot (e.g. 117 117 + // "test.yml"); ServeMux path patterns match a single segment, so the 118 118 + // literal value flows straight through r.PathValue. 119 119 + // 120 120 + // Wire shape per frame: a single TextMessage carrying one JSON 121 121 + // LogLine record (defined in provider.go; byte-compatible with 122 122 + // tangled.org/core/spindle/models.LogLine). The Provider hands us a 123 123 + // channel of LogLine values; we marshal each one and forward it as 124 124 + // one frame so the appview's per-line decode path works unchanged. 125 125 + // 126 126 + // Error mapping is intentionally done *before* the WebSocket upgrade: 127 127 + // ErrLogsNotFound becomes 404 and any other Logs() error becomes 500 128 128 + // so the appview's websocket.DefaultDialer surfaces a real HTTP 129 129 + // status rather than an immediate close. 130 130 + func logsHandler(logger *slog.Logger, provider Provider) http.HandlerFunc { 131 131 + upgrader := websocket.Upgrader{ 132 132 + ReadBufferSize: 1024, 133 133 + WriteBufferSize: 1024, 134 134 + } 135 135 + return func(w http.ResponseWriter, r *http.Request) { 136 136 + knot := r.PathValue("knot") 137 137 + pipelineRkey := r.PathValue("pipelineRkey") 138 138 + workflow := r.PathValue("workflow") 139 139 + 140 140 + // Defensive: ServeMux won't match an empty segment, but a 141 141 + // future router change shouldn't be allowed to silently 142 142 + // produce an "all logs" query. 143 143 + if knot == "" || pipelineRkey == "" || workflow == "" { 144 144 + http.Error(w, "missing path component", http.StatusBadRequest) 145 145 + return 146 146 + } 147 147 + 148 148 + // Establish the log channel BEFORE the WebSocket upgrade so 149 149 + // ErrLogsNotFound / backend errors surface as a real HTTP 150 150 + // status to the appview's dialer rather than as an immediate 151 151 + // post-upgrade close. ctx scopes the producer's lifetime — 152 152 + // it's cancelled below the moment the client disconnects. 153 153 + ctx, cancel := context.WithCancel(r.Context()) 154 154 + defer cancel() 155 155 + 156 156 + ch, err := provider.Logs(ctx, knot, pipelineRkey, workflow) 157 157 + if err != nil { 158 158 + if errors.Is(err, ErrLogsNotFound) { 159 159 + http.Error(w, "logs not found", http.StatusNotFound) 160 160 + return 161 161 + } 162 162 + logger.Error("logs fetch failed", 163 163 + "err", err, 164 164 + "knot", knot, 165 165 + "pipeline_rkey", pipelineRkey, 166 166 + "workflow", workflow, 167 167 + ) 168 168 + http.Error(w, "logs unavailable", http.StatusInternalServerError) 169 169 + return 170 170 + } 171 171 + 172 172 + conn, err := upgrader.Upgrade(w, r, nil) 173 173 + if err != nil { 174 174 + // Upgrade already wrote a response; just record the 175 175 + // failure for diagnostics. 176 176 + logger.Error("logs websocket upgrade failed", 177 177 + "err", err, 178 178 + "knot", knot, 179 179 + "pipeline_rkey", pipelineRkey, 180 180 + "workflow", workflow, 181 181 + ) 182 182 + return 183 183 + } 184 184 + defer func() { 185 185 + // Send a close frame on the way out so the appview proxy 186 186 + // sees a clean shutdown. Mirrors upstream 187 187 + // spindle.(*Spindle).Logs. 188 188 + _ = conn.WriteControl( 189 189 + websocket.CloseMessage, 190 190 + websocket.FormatCloseMessage( 191 191 + websocket.CloseNormalClosure, "log stream complete", 192 192 + ), 193 193 + time.Now().Add(time.Second), 194 194 + ) 195 195 + conn.Close() 196 196 + }() 197 197 + 198 198 + // Detect client disconnect by trying to read; we don't expect 199 199 + // any payloads from the client, so any read outcome (including 200 200 + // EOF) signals the connection has gone away. The cancel hits 201 201 + // the producer goroutine inside the Provider, which stops 202 202 + // sending and closes ch — our drain loop then exits cleanly. 203 203 + go func() { 204 204 + for { 205 205 + if _, _, err := conn.NextReader(); err != nil { 206 206 + cancel() 207 207 + return 208 208 + } 209 209 + } 210 210 + }() 211 211 + 212 212 + // Drain the channel; closure means the run is complete (or 213 213 + // the producer hit ctx). Marshal-then-write each LogLine as 214 214 + // a single TextMessage frame. 215 215 + for { 216 216 + select { 217 217 + case <-ctx.Done(): 218 218 + return 219 219 + case line, ok := <-ch: 220 220 + if !ok { 221 221 + return 222 222 + } 223 223 + frame, err := json.Marshal(line) 224 224 + if err != nil { 225 225 + // The struct is fully internal; a marshal failure 226 226 + // is a programmer bug. Log and bail rather than 227 227 + // poison the stream with a half-frame. 228 228 + logger.Error("marshal log line", 229 229 + "err", err, 230 230 + "knot", knot, 231 231 + "pipeline_rkey", pipelineRkey, 232 232 + "workflow", workflow, 233 233 + ) 234 234 + return 235 235 + } 236 236 + if err := conn.WriteMessage(websocket.TextMessage, frame); err != nil { 237 237 + logger.Debug("logs frame write failed", 238 238 + "err", err, 239 239 + "knot", knot, 240 240 + "pipeline_rkey", pipelineRkey, 241 241 + "workflow", workflow, 242 242 + ) 243 243 + return 244 244 + } 245 245 + } 246 246 + } 100 247 } 101 248 } 102 249

+1 -1

main.go

reviewed

··· 143 143 144 144 // Run the HTTP server. This blocks until ctx is cancelled or the 145 145 // listener errors. 146 146 - if err := runHTTP(ctx, cfg, br); err != nil { 146 146 + if err := runHTTP(ctx, cfg, br, provider); err != nil { 147 147 logger.Error("http server error", "err", err) 148 148 os.Exit(1) 149 149 }

+75 -1

provider.go

reviewed

··· 7 7 8 8 import ( 9 9 "context" 10 10 + "errors" 11 11 + "time" 10 12 11 13 "tangled.org/core/api/tangled" 12 14 ) 13 15 16 16 + // LogLine is the on-the-wire shape of a single log frame emitted by a 17 17 + // Provider. It mirrors tangled.org/core/spindle/models.LogLine on the 18 18 + // JSON level without importing the upstream package — that package 19 19 + // transitively pulls in git, vault, redis and a few hundred other 20 20 + // modules just to expose a handful of types we don't otherwise need. 21 21 + // 22 22 + // The JSON tags here MUST stay byte-compatible with the upstream 23 23 + // struct: the appview's log proxy decodes against the upstream type, 24 24 + // so any tag drift breaks the appview's renderer. 25 25 + type LogLine struct { 26 26 + Kind string `json:"kind"` 27 27 + Content string `json:"content"` 28 28 + Time time.Time `json:"time"` 29 29 + StepId int `json:"step_id"` 30 30 + Stream string `json:"stream,omitempty"` 31 31 + StepStatus string `json:"step_status,omitempty"` 32 32 + StepKind string `json:"step_kind,omitempty"` 33 33 + StepCommand string `json:"step_command,omitempty"` 34 34 + } 35 35 + 36 36 + // LogKind / StepStatus enum values, matching the upstream constants 37 37 + // (LogKindData, LogKindControl, StepStatusStart, StepStatusEnd) on 38 38 + // the wire. Use these instead of bare strings so we don't drift. 39 39 + const ( 40 40 + LogKindData = "data" 41 41 + LogKindControl = "control" 42 42 + StepStatusStart = "start" 43 43 + StepStatusEnd = "end" 44 44 + ) 45 45 + 46 46 + // ErrLogsNotFound is returned by Provider.Logs when the requested 47 47 + // (knot, pipelineRkey, workflow) tuple has no recorded logs — either 48 48 + // because that workflow never ran on this spindle, or because the 49 49 + // provider has since dropped it. The HTTP /logs handler translates 50 50 + // this into a 404 *before* upgrading to WebSocket so the appview's 51 51 + // dialer sees a real HTTP error rather than an immediate close. 52 52 + var ErrLogsNotFound = errors.New("logs not found") 53 53 + 14 54 // Provider dispatches a Tangled pipeline trigger to whatever backend 15 15 - // actually runs the workflows. 55 55 + // actually runs the workflows, and exposes per-workflow log retrieval. 16 56 // 17 57 // Implementations are responsible for publishing 18 58 // sh.tangled.pipeline.status records back through whatever channel ··· 43 83 pipelineRkey string, 44 84 workflows []*tangled.Pipeline_Workflow, 45 85 ) 86 86 + 87 87 + // Logs returns a channel of log frames for a single workflow run, 88 88 + // identified by the same (knot, pipelineRkey, workflow) tuple 89 89 + // Spawn was invoked with. 90 90 + // 91 91 + // Each LogLine corresponds 1:1 to one frame written by the HTTP 92 92 + // /logs WebSocket handler — the handler marshals the LogLine and 93 93 + // emits it as a single TextMessage so appview clients see the 94 94 + // exact same wire shape they would from the stock Tangled spindle 95 95 + // (whose on-disk log file holds the same records, one per line). 96 96 + // 97 97 + // The returned channel is closed by the provider when the stream 98 98 + // is complete; a closed channel is the only "end of stream" 99 99 + // signal — there is no separate done channel. Implementations 100 100 + // MUST also stop sending and close the channel when ctx is 101 101 + // cancelled, so a disconnecting client doesn't leak a producer 102 102 + // goroutine. 103 103 + // 104 104 + // Backend errors that occur *after* a successful return (e.g. 105 105 + // the upstream Buildkite log socket dying mid-stream) are logged 106 106 + // internally and surface to the consumer as an early channel 107 107 + // close. The same observable behaviour the appview already 108 108 + // handles for a websocket that closes mid-stream. 109 109 + // 110 110 + // Returns ErrLogsNotFound if no logs exist for the requested 111 111 + // tuple. Any other returned error indicates a backend failure 112 112 + // and should surface as a 5xx to the HTTP caller. On a non-nil 113 113 + // error, the channel is nil and there is nothing to drain. 114 114 + Logs( 115 115 + ctx context.Context, 116 116 + knot string, 117 117 + pipelineRkey string, 118 118 + workflow string, 119 119 + ) (<-chan LogLine, error) 46 120 }

+122 -8

provider_fake.go

reviewed

··· 5 5 // spawns a goroutine that emits a fixed-cadence stream of 6 6 // sh.tangled.pipeline.status records — "running" every five seconds 7 7 // for thirty seconds, then a final "success" — through the broker. 8 8 + // In parallel it appends synthetic LogLine records into an in-memory 9 9 + // buffer, so /logs queries against a fake run replay something that 10 10 + // matches the upstream spindle's wire format. 8 11 // 9 12 // The point is to exercise the entire trigger → broker → /events → 10 13 // appview path end-to-end before any real CI integration exists. Once ··· 18 21 "encoding/json" 19 22 "fmt" 20 23 "log/slog" 24 24 + "sync" 21 25 "time" 22 26 23 27 "tangled.org/core/api/tangled" ··· 32 36 fakeJobInterval = 5 * time.Second 33 37 ) 34 38 39 39 + // fakeLogChanBuffer is the size of the buffered channel returned by 40 40 + // fakeProvider.Logs. Big enough that a snapshot of a completed fake 41 41 + // run (≈ 8 lines today) drops in without blocking the producer 42 42 + // goroutine even if the consumer is briefly slow; small enough that 43 43 + // a misbehaving consumer can't pin a runaway amount of memory. 44 44 + const fakeLogChanBuffer = 64 45 45 + 35 46 // fakeProvider implements Provider against the in-process broker. 47 47 + // 48 48 + // logs holds the captured per-workflow LogLine slices, keyed by the 49 49 + // fakeLogKey of (knot, pipelineRkey, workflow). The mutex guards the 50 50 + // map *and* concurrent appends to any contained slice — runWorkflow 51 51 + // is the only writer for a given key, but Logs() may snapshot the 52 52 + // slice concurrently. 36 53 type fakeProvider struct { 37 54 br *broker 38 55 log *slog.Logger 56 56 + 57 57 + mu sync.Mutex 58 58 + logs map[string][]LogLine 39 59 } 40 60 41 61 // Compile-time interface check — keeps the fake honest if Provider ··· 47 67 // apart from the knot-consumer / jetstream noise. 48 68 func newFakeProvider(br *broker, log *slog.Logger) *fakeProvider { 49 69 return &fakeProvider{ 50 50 - br: br, 51 51 - log: log.With("component", "provider", "kind", "fake"), 70 70 + br: br, 71 71 + log: log.With("component", "provider", "kind", "fake"), 72 72 + logs: make(map[string][]LogLine), 52 73 } 53 74 } 54 75 ··· 83 104 } 84 105 } 85 106 107 107 + // Logs satisfies Provider. It snapshots the in-memory LogLine slice 108 108 + // for (knot, pipelineRkey, workflow) and returns a buffered channel 109 109 + // that a goroutine drains the snapshot into. The snapshot is taken 110 110 + // under the mutex so subsequent appends by a still-running workflow 111 111 + // won't be reflected — clients that want live updates re-poll. We 112 112 + // don't follow live writes here; the fake's purpose is to exercise 113 113 + // the wire format end-to-end, not to reproduce hpcloud/tail. 114 114 + // 115 115 + // The producer goroutine honours ctx so that a disconnecting HTTP 116 116 + // client tears the channel send down promptly instead of waiting on 117 117 + // an unbuffered receiver. 118 118 + func (p *fakeProvider) Logs( 119 119 + ctx context.Context, 120 120 + knot string, 121 121 + pipelineRkey string, 122 122 + workflow string, 123 123 + ) (<-chan LogLine, error) { 124 124 + key := fakeLogKey(knot, pipelineRkey, workflow) 125 125 + 126 126 + p.mu.Lock() 127 127 + stored, ok := p.logs[key] 128 128 + if !ok { 129 129 + p.mu.Unlock() 130 130 + return nil, ErrLogsNotFound 131 131 + } 132 132 + // Copy under the lock so subsequent appendLogLine writes can't 133 133 + // race with our drain goroutine reading the slice header. 134 134 + snapshot := make([]LogLine, len(stored)) 135 135 + copy(snapshot, stored) 136 136 + p.mu.Unlock() 137 137 + 138 138 + out := make(chan LogLine, fakeLogChanBuffer) 139 139 + go func() { 140 140 + defer close(out) 141 141 + for _, line := range snapshot { 142 142 + select { 143 143 + case <-ctx.Done(): 144 144 + return 145 145 + case out <- line: 146 146 + } 147 147 + } 148 148 + }() 149 149 + return out, nil 150 150 + } 151 151 + 86 152 // runWorkflow emits a "running" status every fakeJobInterval until 87 87 - // fakeJobDuration elapses, then a final "success". On ctx 88 88 - // cancellation it returns without issuing the terminal publish — the 89 89 - // broker's underlying store may already be closing during shutdown. 153 153 + // fakeJobDuration elapses, then a final "success". Alongside each 154 154 + // status it appends a corresponding LogLine into the in-memory log 155 155 + // buffer so /logs returns something coherent. 156 156 + // 157 157 + // On ctx cancellation it returns without issuing the terminal publish 158 158 + // or the closing control frame — the broker's underlying store may 159 159 + // already be closing during shutdown. 90 160 func (p *fakeProvider) runWorkflow(ctx context.Context, knot, pipelineRkey, workflow string) { 91 161 // pipelineURI is what the appview parses out of the status record 92 162 // to associate it back with the originating pipeline. Format ··· 103 173 "workflow", workflow, 104 174 ) 105 175 176 176 + // Start control frame. The appview's log renderer keys timing on 177 177 + // matching start/end control frames per step_id, so we always 178 178 + // emit one even though the fake has only a single synthetic step. 179 179 + p.appendLogLine(knot, pipelineRkey, workflow, LogLine{ 180 180 + Kind: LogKindControl, 181 181 + Time: time.Now(), 182 182 + Content: workflow, 183 183 + StepId: 0, 184 184 + StepStatus: StepStatusStart, 185 185 + }) 186 186 + 106 187 // Heartbeat phase. seq doubles as a per-workflow disambiguator 107 188 // in the synthesized status rkey so concurrent fakes (across 108 189 // workflows or pipelines) don't collide. 109 190 deadline := time.Now().Add(fakeJobDuration) 110 191 seq := 0 111 192 for time.Now().Before(deadline) { 193 193 + p.appendLogLine(knot, pipelineRkey, workflow, LogLine{ 194 194 + Kind: LogKindData, 195 195 + Time: time.Now(), 196 196 + Content: fmt.Sprintf("[fake] heartbeat %d at %s\n", 197 197 + seq, time.Now().UTC().Format(time.RFC3339), 198 198 + ), 199 199 + StepId: 0, 200 200 + Stream: "stdout", 201 201 + }) 112 202 if err := p.publishStatus(ctx, pipelineURI, workflow, "running", seq); err != nil { 113 203 logger.Error("publish fake running status", "err", err, "seq", seq) 114 204 return ··· 122 212 } 123 213 } 124 214 125 125 - // Terminal publish. "success" matches the upstream StatusKind 126 126 - // enum (see tangled.org/core/spindle/models) — the appview 127 127 - // routes status strings through that same enum. 215 215 + // End control + terminal status publish. "success" matches the 216 216 + // upstream StatusKind enum (see tangled.org/core/spindle/models) 217 217 + // — the appview routes status strings through that same enum. 218 218 + p.appendLogLine(knot, pipelineRkey, workflow, LogLine{ 219 219 + Kind: LogKindControl, 220 220 + Time: time.Now(), 221 221 + Content: workflow, 222 222 + StepId: 0, 223 223 + StepStatus: StepStatusEnd, 224 224 + }) 128 225 if err := p.publishStatus(ctx, pipelineURI, workflow, "success", seq); err != nil { 129 226 logger.Error("publish fake success status", "err", err, "seq", seq) 130 227 return ··· 155 252 } 156 253 return nil 157 254 } 255 255 + 256 256 + // appendLogLine records line as the next captured frame for the 257 257 + // workflow, allocating a slice on first use. Holding the mutex across 258 258 + // the append ensures a concurrent Logs() snapshot doesn't observe a 259 259 + // half-grown slice header. 260 260 + func (p *fakeProvider) appendLogLine(knot, pipelineRkey, workflow string, line LogLine) { 261 261 + key := fakeLogKey(knot, pipelineRkey, workflow) 262 262 + p.mu.Lock() 263 263 + defer p.mu.Unlock() 264 264 + p.logs[key] = append(p.logs[key], line) 265 265 + } 266 266 + 267 267 + // fakeLogKey is the canonical map key for the in-memory log buffer. 268 268 + // Centralised so reader and writer can't drift on separator choice. 269 269 + func fakeLogKey(knot, pipelineRkey, workflow string) string { 270 270 + return knot + "\x00" + pipelineRkey + "\x00" + workflow 271 271 + }