fix(host): AppendNow warns when the log offset may be stale after a send cut short (GO-005)

commit: 799f55ca0d6c90831f0eedb1d12f5b9d7716fd6e
author: brice · 9d ago · 2026-10-12 20:13 UTC
parent: ca9abad
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

2 files · +69 -1

M internal/host/outbox.go +24 -1


@@ -62,6 +62,10 @@ type outbox struct {
62	62		done chan struct{}
63	63
64	64		offset int64 // lines the server holds, as far as we know
	65	+	// offsetMaybeStale is set when a send was cut short by its context while
	66	+	// the call may have been applied by the server with the response lost: the
	67	+	// offset is then possibly behind the server's count.
	68	+	offsetMaybeStale bool
65	69		}
66	70
67	71		// callClient is the part of the RunnerService client the job uses.

@@ -277,6 +281,7 @@ func (o *outbox) sendLines(ctx context.Context, batch []event) {
277	281		switch {
278	282		case err == nil:
279	283		o.mu.Lock()
	284	+	o.offsetMaybeStale = false
280	285		// The server says how many lines it holds: that is the truth. A
281	286		// difference means lines were stored that we did not count (a retry
282	287		// whose answer we lost) or dropped; continue from the server's count,

@@ -287,7 +292,13 @@ func (o *outbox) sendLines(ctx context.Context, batch []event) {
287	292		}
288	293		o.offset = next
289	294		o.mu.Unlock()
290		-	case ctx.Err() == nil && classify(err) == kindPermanent:
	295	+	case ctx.Err() != nil:
	296	+	// Cut short in the middle of a call or of its retries: the server may
	297	+	// hold these lines without our having seen the answer.
	298	+	o.mu.Lock()
	299	+	o.offsetMaybeStale = true
	300	+	o.mu.Unlock()
	301	+	case classify(err) == kindPermanent:
291	302		// The server refused these lines; they are dropped and the offset stays
292	303		// where the server is.
293	304		o.log.Error().Err(err).Int("lines", len(req.Lines)).Int32("step", req.Step).Msg("AppendLog refused, the lines are dropped")

@@ -297,6 +308,18 @@ func (o *outbox) sendLines(ctx context.Context, batch []event) {
297	308		// AppendNow sends one line synchronously, after Close: the sender has stopped,
298	309		// so nothing else uses the offset. It is for the last words of a job (an error
299	310		// that explains why the result could not be reported).
	311	+	//
	312	+	// If Close gave up in the middle of a send, the offset may be behind the
	313	+	// server's: the line is sent anyway (the server answers with its own count, and
	314	+	// an offset that is too low makes it skip the line as a repeat or refuse it),
	315	+	// and the warning says why it may be missing.
300	316		func (o *outbox) AppendNow(ctx context.Context, l logLine) {
	317	+	o.mu.Lock()
	318	+	stale := o.offsetMaybeStale
	319	+	offset := o.offset
	320	+	o.mu.Unlock()
	321	+	if stale {
	322	+	o.log.Warn().Int64("offset", offset).Msg("the log offset may be stale: Close was cut short in the middle of a send, so the last line may be skipped or refused by the server")
	323	+	}
301	324		o.sendLines(ctx, []event{{line: &l}})
302	325		}

 	done    chan struct{}
 	offset int64 // lines the server holds, as far as we know
+	// offsetMaybeStale is set when a send was cut short by its context while
+	// the call may have been applied by the server with the response lost: the
+	// offset is then possibly behind the server's count.
+	offsetMaybeStale bool
 }
 // callClient is the part of the RunnerService client the job uses.
 	switch {
 	case err == nil:
 		o.mu.Lock()
+		o.offsetMaybeStale = false
 		// The server says how many lines it holds: that is the truth. A
 		// difference means lines were stored that we did not count (a retry
 		// whose answer we lost) or dropped; continue from the server's count,
 		}
 		o.offset = next
 		o.mu.Unlock()
-	case ctx.Err() == nil && classify(err) == kindPermanent:
+	case ctx.Err() != nil:
+		// Cut short in the middle of a call or of its retries: the server may
+		// hold these lines without our having seen the answer.
+		o.mu.Lock()
+		o.offsetMaybeStale = true
+		o.mu.Unlock()
+	case classify(err) == kindPermanent:
 		// The server refused these lines; they are dropped and the offset stays
 		// where the server is.
 		o.log.Error().Err(err).Int("lines", len(req.Lines)).Int32("step", req.Step).Msg("AppendLog refused, the lines are dropped")
 // AppendNow sends one line synchronously, after Close: the sender has stopped,
 // so nothing else uses the offset. It is for the last words of a job (an error
 // that explains why the result could not be reported).
+//
+// If Close gave up in the middle of a send, the offset may be behind the
+// server's: the line is sent anyway (the server answers with its own count, and
+// an offset that is too low makes it skip the line as a repeat or refuse it),
+// and the warning says why it may be missing.
 func (o *outbox) AppendNow(ctx context.Context, l logLine) {
+	o.mu.Lock()
+	stale := o.offsetMaybeStale
+	offset := o.offset
+	o.mu.Unlock()
+	if stale {
+		o.log.Warn().Int64("offset", offset).Msg("the log offset may be stale: Close was cut short in the middle of a send, so the last line may be skipped or refused by the server")
+	}
 	o.sendLines(ctx, []event{{line: &l}})
 }

A internal/host/outbox_test.go +45 -0

@@ -0,0 +1,45 @@
	1	+	// SPDX-License-Identifier: MIT
	2	+
	3	+	package host
	4	+
	5	+	import (
	6	+	"context"
	7	+	"strings"
	8	+	"testing"
	9	+	"time"
	10	+
	11	+	"github.com/bnema/zerowrap"
	12	+	)
	13	+
	14	+	// AppendNow says so when an earlier send was cut short and the offset may be
	15	+	// behind the server's (GO-005), and stays silent otherwise.
	16	+	func TestAppendNowWarnsOfAStaleOffset(t *testing.T) {
	17	+	const warning = "log offset may be stale"
	18	+	logs := &logBuf{}
	19	+	log := zerowrap.New(zerowrap.Config{Level: "debug", Format: "json", Output: logs})
	20	+	tm := Timing{CallTimeout: time.Hour, BackoffMin: time.Millisecond, BackoffMax: time.Millisecond}
	21	+	cl := &stubClient{hang: true}
	22	+	o := newOutbox(cl, "j1", tm, func(error) {}, log)
	23	+
	24	+	// A send that Close cut short: the server may have applied it.
	25	+	ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond)
	26	+	defer cancel()
	27	+	o.sendLines(ctx, []event{{line: &logLine{step: 1, text: "a"}}})
	28	+
	29	+	cl.mu.Lock()
	30	+	cl.hang = false
	31	+	cl.mu.Unlock()
	32	+	o.AppendNow(context.Background(), logLine{step: 1, text: "b"})
	33	+	if !strings.Contains(logs.String(), warning) {
	34	+	t.Fatalf("no warning about the offset: %q", logs.String())
	35	+	}
	36	+
	37	+	// The send went through: the offset is the server's again.
	38	+	logs.mu.Lock()
	39	+	logs.b.Reset()
	40	+	logs.mu.Unlock()
	41	+	o.AppendNow(context.Background(), logLine{step: 1, text: "c"})
	42	+	if strings.Contains(logs.String(), warning) {
	43	+	t.Fatalf("a warning with a fresh offset: %q", logs.String())
	44	+	}
	45	+	}

@@ -0,0 +1,45 @@
	1	+	// SPDX-License-Identifier: MIT
	2	+
	3	+	package host
	4	+
	5	+	import (
	6	+	"context"
	7	+	"strings"
	8	+	"testing"
	9	+	"time"
	10	+
	11	+	"github.com/bnema/zerowrap"
	12	+	)
	13	+
	14	+	// AppendNow says so when an earlier send was cut short and the offset may be
	15	+	// behind the server's (GO-005), and stays silent otherwise.
	16	+	func TestAppendNowWarnsOfAStaleOffset(t *testing.T) {
	17	+	const warning = "log offset may be stale"
	18	+	logs := &logBuf{}
	19	+	log := zerowrap.New(zerowrap.Config{Level: "debug", Format: "json", Output: logs})
	20	+	tm := Timing{CallTimeout: time.Hour, BackoffMin: time.Millisecond, BackoffMax: time.Millisecond}
	21	+	cl := &stubClient{hang: true}
	22	+	o := newOutbox(cl, "j1", tm, func(error) {}, log)
	23	+
	24	+	// A send that Close cut short: the server may have applied it.
	25	+	ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond)
	26	+	defer cancel()
	27	+	o.sendLines(ctx, []event{{line: &logLine{step: 1, text: "a"}}})
	28	+
	29	+	cl.mu.Lock()
	30	+	cl.hang = false
	31	+	cl.mu.Unlock()
	32	+	o.AppendNow(context.Background(), logLine{step: 1, text: "b"})
	33	+	if !strings.Contains(logs.String(), warning) {
	34	+	t.Fatalf("no warning about the offset: %q", logs.String())
	35	+	}
	36	+
	37	+	// The send went through: the offset is the server's again.
	38	+	logs.mu.Lock()
	39	+	logs.b.Reset()
	40	+	logs.mu.Unlock()
	41	+	o.AppendNow(context.Background(), logLine{step: 1, text: "c"})
	42	+	if strings.Contains(logs.String(), warning) {
	43	+	t.Fatalf("a warning with a fresh offset: %q", logs.String())
	44	+	}
	45	+	}