package plan
import (
"bufio"
"context"
"encoding/json"
"fmt"
"io"
"os"
"os/exec"
"sync"
)
// BuildState is one package currently being worked on.
type BuildState struct {
Name string `json:"name"`
Phase string `json:"phase"`
Elapsed float64 `json:"elapsed"`
Jobs int `json:"jobs"`
//: what the jobs are, e.g. "rustc ×4, cc1". A count says something is
//: happening; this says what, which is what tells a compiling build from
//: a wedged one.
Workers string `json:"workers"`
}
// Event is one line of the backend's protocol.
//
// One struct rather than a type per event: the fields present depend on Event,
// and a struct keeps the reader trivial. Unknown fields are ignored, so the
// backend can add to the protocol without breaking an older front end.
type Event struct {
Event string `json:"event"`
// start / phase_start
Phase string `json:"phase"`
Command string `json:"command"`
Log string `json:"log"`
Phases []string `json:"phases"`
Total int `json:"total"`
//: the run's total in work rather than in packages, plus the per-package
//: weights, listed only for those worth more than one. A build runs for
//: minutes and a download for about a second; counted alike, the bar raced
//: to nearly full during the repository phase.
Weight int `json:"weight"`
Weights map[string]int `json:"weights"`
Held []string `json:"held"`
Skipped []string `json:"skipped"`
// error
Details []string `json:"details"`
// build
Packages []BuildState `json:"packages"`
LogLines []string `json:"log_lines"`
DownloadBytes int64 `json:"download_bytes"`
DownloadRate float64 `json:"download_rate"`
Done int `json:"done"`
// finished
Name string `json:"name"`
Elapsed float64 `json:"elapsed"`
State string `json:"state"`
Version string `json:"version"`
Shared bool `json:"shared"`
// phase_end / done
Returncode int `json:"returncode"`
Failed []string `json:"failed"`
Message string `json:"message"`
// recovery: a way to finish a run that could not proceed unattended.
// Its own key, not `command`: that one is already a string on the phase
// events, and two fields sharing a JSON tag silently breaks the decoder.
RecoveryLabel string `json:"label"`
RecoveryCommand []string `json:"argv"`
RecoveryNote string `json:"note"`
//: why a package failed. Its own key, because `note` already belongs to
//: the recovery event -- one tag, two meanings, and the decoder silently
//: filled the wrong field.
FailureReason string `json:"reason"`
}
// Stream is a running apply, read from a channel.
type Stream struct {
Events <-chan Event
//: ProtocolErrors reports lines the reader could not parse. Surfaced rather
//: than swallowed: a protocol break that nobody notices is worse than a
//: crash, because everything downstream quietly renders stale state.
ProtocolErrors <-chan error
//: Wait blocks until the process exits and returns its status.
Wait func() error
//: Stderr captures the backend's human output, for error reporting.
Stderr func() string
//: Abort asks the backend to stop.
Abort func()
//: Skip asks it to abandon the package in progress and carry on. Only the
//: AUR phase has a package in progress to abandon: the others run as one
//: transaction, and nothing there can be skipped without unpicking it.
Skip func()
}
// StartApply launches `updater apply --json-stream` and streams its events.
//
// The backend writes JSON and nothing else on stdout; anything human goes to
// stderr. A line that does not parse is surfaced rather than skipped, because
// silently dropping it would hide a protocol break.
func (b Backend) StartApply(ctx context.Context, skips, holds []string) (*Stream, error) {
args := []string{"apply", "--json-stream", "--yes"}
for _, s := range skips {
args = append(args, "--skip", s)
}
for _, h := range holds {
args = append(args, "--hold", h)
}
args = append(args, b.Args...)
full := append([]string{"-m", b.Module}, args...)
cmd := exec.CommandContext(ctx, b.Python, full...)
cmd.Env = append(os.Environ(), b.Env...)
stdout, err := cmd.StdoutPipe()
if err != nil {
return nil, err
}
var stderrBuf lockedBuffer
cmd.Stderr = &stderrBuf
if err := cmd.Start(); err != nil {
return nil, &Error{Args: append([]string{b.Python}, full...), Err: err}
}
events := make(chan Event, 64)
protocolErrors := make(chan error, 8)
go func() {
defer close(events)
defer close(protocolErrors)
scanner := bufio.NewScanner(stdout)
// a diff or a long command line can exceed the default 64 KiB
scanner.Buffer(make([]byte, 0, 64*1024), 4*1024*1024)
for scanner.Scan() {
line := scanner.Bytes()
if len(line) == 0 {
continue
}
var event Event
if err := json.Unmarshal(line, &event); err != nil {
select {
case protocolErrors <- fmt.Errorf(
"backend sent a line that is not JSON: %q", truncate(string(line), 200)):
default: // nobody reading; do not stall the stream over it
}
continue
}
events <- event
}
if err := scanner.Err(); err != nil && err != io.EOF {
select {
case protocolErrors <- err:
default:
}
}
}()
return &Stream{
Events: events,
ProtocolErrors: protocolErrors,
Wait: func() error { return cmd.Wait() },
Stderr: stderrBuf.String,
Abort: func() {
if cmd.Process != nil {
// SIGINT, not SIGKILL: pacman and makepkg unwind a half-finished
// transaction on it, which is the whole point
_ = cmd.Process.Signal(interruptSignal())
}
},
Skip: func() {
if cmd.Process != nil {
// the backend alone, not its group: the child is in its own
// group and must not see this
_ = cmd.Process.Signal(skipSignal())
}
},
}, nil
}
// lockedBuffer is a minimal concurrency-safe buffer for the child's stderr.
type lockedBuffer struct {
mu sync.Mutex
buf []byte
}
func (b *lockedBuffer) Write(p []byte) (int, error) {
b.mu.Lock()
defer b.mu.Unlock()
b.buf = append(b.buf, p...)
if len(b.buf) > 64*1024 {
b.buf = b.buf[len(b.buf)-64*1024:]
}
return len(p), nil
}
func (b *lockedBuffer) String() string {
b.mu.Lock()
defer b.mu.Unlock()
return string(b.buf)
}
func truncate(s string, n int) string {
if len(s) <= n {
return s
}
return s[:n] + "…"
}