Bug 14: stream journal reads instead of buffering them; soft memory limit for the GC

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

internal/cron/cron.go

@@ -116,6 +116,35 @@ func run(ctx context.Context, stdin string, args ...string) (string, error) {
 	return out.String(), nil
 }
 
+// runLines streams a command's stdout to fn line by line, so multi-megabyte journal
+// dumps never sit in memory whole.
+func runLines(ctx context.Context, fn func(line []byte), args ...string) error {
+	ctx, cancel := context.WithTimeout(ctx, 30*time.Second)
+	defer cancel()
+	cmd := exec.CommandContext(ctx, args[0], args[1:]...)
+	var errb bytes.Buffer
+	cmd.Stderr = &errb
+	stdout, err := cmd.StdoutPipe()
+	if err != nil {
+		return err
+	}
+	if err := cmd.Start(); err != nil {
+		return err
+	}
+	sc := bufio.NewScanner(stdout)
+	sc.Buffer(make([]byte, 0, 64<<10), 4<<20)
+	for sc.Scan() {
+		fn(sc.Bytes())
+	}
+	if err := cmd.Wait(); err != nil {
+		if msg := strings.TrimSpace(errb.String()); msg != "" {
+			return errors.New(msg)
+		}
+		return err
+	}
+	return nil
+}
+
 func (m *Manager) sudoArgs(args ...string) []string {
 	if m.sudo {
 		return append([]string{"sudo", "-n"}, args...)
@@ -414,18 +443,15 @@ func (m *Manager) lastRunsFromJournal(ctx context.Context) map[string]int64 {
 	if m.journalSudo {
 		args = append([]string{"sudo", "-n"}, args...)
 	}
-	out, _ := run(ctx, "", args...)
 	seen := m.logging
-	sc := bufio.NewScanner(strings.NewReader(out))
-	sc.Buffer(make([]byte, 0, 64<<10), 4<<20)
-	for sc.Scan() {
+	_ = runLines(ctx, func(line []byte) {
 		var raw struct {
 			Msg    any    `json:"MESSAGE"`
 			TS     string `json:"__REALTIME_TIMESTAMP"`
 			Cursor string `json:"__CURSOR"`
 		}
-		if json.Unmarshal(sc.Bytes(), &raw) != nil {
-			continue
+		if json.Unmarshal(line, &raw) != nil {
+			return
 		}
 		seen = true
 		if raw.Cursor != "" {
@@ -434,14 +460,14 @@ func (m *Manager) lastRunsFromJournal(ctx context.Context) map[string]int64 {
 		msg, _ := raw.Msg.(string)
 		mm := cronLogRe.FindStringSubmatch(msg)
 		if mm == nil {
-			continue
+			return
 		}
 		us, _ := strconv.ParseInt(raw.TS, 10, 64)
 		k := mm[1] + "\x00" + mm[2]
 		if us/1000 > res[k] {
 			res[k] = us / 1000
 		}
-	}
+	}, args...)
 	m.mu.Lock()
 	m.lastRuns, m.lastAt, m.lastCursor, m.logging = res, time.Now(), cursor, seen
 	m.mu.Unlock()
@@ -454,28 +480,29 @@ func (m *Manager) JobLogs(ctx context.Context, j Job, n int) ([]map[string]any,
 	if m.journalSudo {
 		args = append([]string{"sudo", "-n"}, args...)
 	}
-	out, err := run(ctx, "", args...)
-	if err != nil {
-		return nil, err
-	}
 	var res []map[string]any
-	sc := bufio.NewScanner(strings.NewReader(out))
-	sc.Buffer(make([]byte, 0, 64<<10), 4<<20)
-	for sc.Scan() {
+	needle := "(" + j.User + ") CMD (" + j.Command + ")"
+	err := runLines(ctx, func(line []byte) {
+		if !bytes.Contains(line, []byte(j.Command[:min(len(j.Command), 40)])) { // cheap pre-filter before JSON
+			return
+		}
 		var raw struct {
 			Msg any    `json:"MESSAGE"`
 			TS  string `json:"__REALTIME_TIMESTAMP"`
 			PID string `json:"_PID"`
 		}
-		if json.Unmarshal(sc.Bytes(), &raw) != nil {
-			continue
+		if json.Unmarshal(line, &raw) != nil {
+			return
 		}
 		msg, _ := raw.Msg.(string)
-		if !strings.Contains(msg, "("+j.User+") CMD ("+j.Command+")") {
-			continue
+		if !strings.Contains(msg, needle) {
+			return
 		}
 		us, _ := strconv.ParseInt(raw.TS, 10, 64)
 		res = append(res, map[string]any{"t": us / 1000, "p": 6, "ident": "CRON", "pid": raw.PID, "msg": msg})
+	}, args...)
+	if err != nil {
+		return nil, err
 	}
 	if len(res) > n {
 		res = res[len(res)-n:]