Streamline run log and update handling

runtime/lifecycle.go

@@ -32,6 +32,7 @@ const (
 	defaultLifecycleTimeout = 30 * time.Second
 	maxLifecycleTimeout     = 2 * time.Hour
 	maxLifecycleOutputBytes = 4096
+	maxLifecycleLogChunk    = 64 * 1024
 )
 
 var (
@@ -960,10 +961,16 @@ func (supervisor OSProcessSupervisor) Run(ctx context.Context, command ProcessCo
 	for key, value := range command.Env {
 		cmd.Env = append(cmd.Env, key+"="+value)
 	}
+	stdoutLimit := maxLifecycleOutputBytes
+	stderrLimit := maxLifecycleOutputBytes
+	if command.OutputLine != nil {
+		stdoutLimit = 0
+		stderrLimit = 0
+	}
 	var stdout bytes.Buffer
 	var stderr bytes.Buffer
-	stdoutWriter := newLifecycleOutputWriter(&stdout, maxLifecycleOutputBytes, command, "stdout")
-	stderrWriter := newLifecycleOutputWriter(&stderr, maxLifecycleOutputBytes, command, "stderr")
+	stdoutWriter := newLifecycleOutputWriter(&stdout, stdoutLimit, command, "stdout")
+	stderrWriter := newLifecycleOutputWriter(&stderr, stderrLimit, command, "stderr")
 	cmd.Stdout = stdoutWriter
 	cmd.Stderr = stderrWriter
 	if err := cmd.Start(); err != nil {
@@ -1014,6 +1021,9 @@ func (writer *lifecycleOutputWriter) Write(p []byte) (int, error) {
 			_, _ = writer.buffer.Write(p)
 		}
 	}
+	if writer.command.OutputLine == nil {
+		return len(p), nil
+	}
 	writer.pending += string(p)
 	for {
 		index := strings.IndexByte(writer.pending, '\n')
@@ -1024,6 +1034,10 @@ func (writer *lifecycleOutputWriter) Write(p []byte) (int, error) {
 		writer.pending = writer.pending[index+1:]
 		writer.logLine(line)
 	}
+	for len(writer.pending) >= maxLifecycleLogChunk {
+		writer.logLine(writer.pending[:maxLifecycleLogChunk])
+		writer.pending = writer.pending[maxLifecycleLogChunk:]
+	}
 	return len(p), nil
 }
 

runtime/lifecycle_test.go

@@ -553,6 +553,27 @@ func TestOSProcessSupervisorRelaysRawCarriageReturn(t *testing.T) {
 	if len(output) != 1 || output[0] != "stdout:raw\r" {
 		t.Fatalf("expected raw CR to be relayed, got %+v", output)
 	}
+	if result.Stdout != "" || result.Stderr != "" {
+		t.Fatalf("relayed process output must not be duplicated into result buffers, got %+v", result)
+	}
+}
+
+func TestOSProcessSupervisorChunksLongRelayedLinesWithoutRewriting(t *testing.T) {
+	var output []string
+	result, err := (OSProcessSupervisor{}).Run(context.Background(), ProcessCommand{
+		Args: []string{"sh", "-c", "printf %70000s | tr ' ' x"},
+		OutputLine: func(stream string, line string) {
+			if stream == "stdout" {
+				output = append(output, line)
+			}
+		},
+	})
+	if err != nil || result.ExitCode != 0 {
+		t.Fatalf("run process: result=%+v err=%v", result, err)
+	}
+	if len(output) < 2 || strings.Join(output, "") != strings.Repeat("x", 70000) {
+		t.Fatalf("expected long relayed line to reassemble exactly, chunks=%d", len(output))
+	}
 }
 
 func TestProcessCommandJSONOmitsOutputCallback(t *testing.T) {

runtime/self_update.go

@@ -27,6 +27,7 @@ const (
 	maxSelfUpdateBytes        = int64(512 * 1024 * 1024)
 	maxSelfUpdateEntries      = 8
 	defaultUpdateHealthWait   = 30 * time.Second
+	selfUpdateProgressEvery   = int64(8 * 1024 * 1024)
 )
 
 var ErrSelfUpdateRestartRequested = errors.New("Run self-update restart requested")
@@ -217,6 +218,20 @@ func (worker *Worker) downloadRunUpdate(ctx context.Context, assignment protocol
 	if _, err := file.Seek(offset, io.SeekStart); err != nil {
 		return err
 	}
+	manifest.DownloadedBytes = offset
+	lastPersistedOffset := offset
+	persistProgress := func(force bool) error {
+		if !force && offset < input.SizeBytes && offset-lastPersistedOffset < selfUpdateProgressEvery {
+			return nil
+		}
+		manifest.DownloadedBytes = offset
+		manifest.UpdatedAt = time.Now().UTC()
+		if err := persistSelfUpdateManifest(manifestPath, *manifest); err != nil {
+			return err
+		}
+		lastPersistedOffset = offset
+		return nil
+	}
 	for offset < input.SizeBytes {
 		if ctx.Err() != nil {
 			return ctx.Err()
@@ -239,16 +254,17 @@ func (worker *Worker) downloadRunUpdate(ctx context.Context, assignment protocol
 		if _, err := file.Write(chunk.Payload); err != nil {
 			return err
 		}
+		offset += int64(len(chunk.Payload))
+		if err := persistProgress(false); err != nil {
+			return err
+		}
+	}
 	if err := file.Sync(); err != nil {
 		return err
 	}
-		offset += int64(len(chunk.Payload))
-		manifest.DownloadedBytes = offset
-		manifest.UpdatedAt = time.Now().UTC()
-		if err := persistSelfUpdateManifest(manifestPath, *manifest); err != nil {
+	if err := persistProgress(true); err != nil {
 		return err
 	}
-	}
 	if err := file.Close(); err != nil {
 		return err
 	}