Let dev builds shorten the connector's handoff grace, and test requests from before a restart

e2e/connect/grace_test.go

@@ -0,0 +1,108 @@
+//go:build linux || darwin
+
+package connect
+
+import (
+	"encoding/json"
+	"strconv"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/basecamp/basecamp-cli/internal/connector/fakebasecamp"
+)
+
+// The handoff grace the after-grace test runs with, and how long before it
+// is emitted the fake says its request was made. The age is past the grace
+// and well inside connector.HandoffGrace, the minute a release build waits:
+// the request is discarded only because the shortened grace is in effect.
+const (
+	shortGrace = 2 * time.Second
+	requestAge = 10 * time.Second
+)
+
+// MsgBeforeThisRun is logged when the handoff closes a request made before
+// the run's grace began, without writing it.
+const MsgBeforeThisRun = "connector: a request from before this run started is not handed off"
+
+// The connector admits a request but crashes before it is handed off: here
+// it is held with --hold, so nothing is handed off, and killed outright.
+// The operator releases the hold. The connector starts again only after the
+// grace has passed since the request was made, and by then nobody is there
+// to take it: the request is closed as before_this_run, and no request line
+// is written for it.
+//
+// The fake stamps the request requestAge before it is emitted, rather than
+// the test waiting out the grace. That is the same request to the
+// connector: whether it hands a record off depends only on the event's
+// created_at, as the feed served it, against when the run started
+// (handOff in internal/connector/handoff.go), and nothing else in intake or
+// admission reads created_at. The age exceeds the grace on the restart
+// however fast the steps between run, and no step waits on a timer.
+//
+// Within the grace, the same steps hand the request off; that is the
+// knob-free failure scenarios' test, under the default grace.
+func TestARequestMadeBeforeTheGraceIsNotHandedOffOnARestart(t *testing.T) {
+	if !devBuild {
+		t.Skip("the handoff grace can be shortened only in a dev build (go test -tags dev, as make test and CI run it); a release build reads no environment for it")
+	}
+	t.Parallel()
+	h := NewHarness(t, world(), fakebasecamp.WithClock(func() time.Time { return time.Now().Add(-requestAge) }))
+	h.env = append(h.env, "BASECAMP_CONNECT_HANDOFF_GRACE="+shortGrace.String())
+	h.Setup(t)
+
+	held := h.Connect(t, "--hold")
+	held.WaitStreaming(t, 1)
+	ev := comment(h, fakebasecamp.OperatorID, askAgent(), fakebasecamp.Live)
+	out := held.WaitFor(t, "the event line", func(o Output) bool { return len(o.EventsFor(ev.ID)) > 0 })
+	events := out.EventsFor(ev.ID)
+	require.Len(t, events, 1)
+	require.Equal(t, "admitted", events[0].State, "a trusted mention is admitted under the hold")
+	held.Kill(t)
+	out = held.Wait(t)
+	require.Empty(t, out.RequestsFor(ev.ID), "nothing is handed off under the hold")
+
+	h.MustRun(t, "connect", "release", "-P", Agent)
+
+	// The request's age as the connector will judge it, from the created_at
+	// the feed served: it is already past the grace, before the restart.
+	pointers := out.PointersFor(ev.ID)
+	require.Len(t, pointers, 1)
+	createdAt, err := time.Parse(time.RFC3339Nano, pointers[0].CreatedAt)
+	require.NoError(t, err)
+	require.Greater(t, time.Since(createdAt), shortGrace, "the request is older than the grace before the connector starts again")
+
+	restarted := h.Connect(t)
+	id := strconv.FormatInt(ev.ID, 10)
+	restarted.WaitFor(t, "the request closed as before this run", func(o Output) bool {
+		for _, l := range o.Logged(MsgBeforeThisRun) {
+			if l.Attrs["event_id"] == id {
+				return true
+			}
+		}
+		return false
+	})
+	out = restarted.Stop(t)
+
+	assert.Empty(t, out.Requests(), "no request line on the restart")
+	overridden := out.Logged("connector: the handoff grace is overridden for this dev build")
+	require.Len(t, overridden, 1)
+	assert.Equal(t, shortGrace.String(), overridden[0].Attrs["grace"])
+
+	// The ledger agrees: the record is closed, so no later run hands it
+	// off either. The connector logs the close only once it is written.
+	res := h.MustRun(t, "connect", "status", "-P", Agent, "--json")
+	var status struct {
+		Data struct {
+			Status struct {
+				Queues map[string]int `json:"queues"`
+			} `json:"status"`
+		} `json:"data"`
+	}
+	require.NoError(t, json.Unmarshal([]byte(res.Stdout), &status), res.Stdout)
+	require.NotNil(t, status.Data.Status.Queues, "status reports its queues\n%s", res.Stdout)
+	assert.Zero(t, status.Data.Status.Queues["admitted"], res.Stdout)
+	assert.Zero(t, status.Data.Status.Queues["queued"], res.Stdout)
+}

internal/commands/connect_handoff_grace_dev.go

@@ -0,0 +1,37 @@
+//go:build dev
+
+package commands
+
+import (
+	"fmt"
+	"os"
+	"time"
+)
+
+// connectHandoffGraceEnv sets, in a dev build only, how long before the run
+// started a request may have been made and still be handed off, as a Go
+// duration. The end-to-end tests shorten it, so a run that starts after the
+// grace has passed is one they can reach in seconds rather than a minute. A
+// release build reads no environment for it at all: see
+// connect_handoff_grace_stub.go.
+const connectHandoffGraceEnv = "BASECAMP_CONNECT_HANDOFF_GRACE"
+
+// connectHandoffGrace is the grace connectHandoffGraceEnv sets, zero when
+// it is unset, so the handoff's own default applies.
+func connectHandoffGrace() (time.Duration, error) {
+	return parseConnectHandoffGrace(os.Getenv(connectHandoffGraceEnv))
+}
+
+func parseConnectHandoffGrace(v string) (time.Duration, error) {
+	if v == "" {
+		return 0, nil
+	}
+	d, err := time.ParseDuration(v)
+	if err != nil {
+		return 0, fmt.Errorf("%s is not a duration: %w", connectHandoffGraceEnv, err)
+	}
+	if d <= 0 {
+		return 0, fmt.Errorf("%s must be a positive duration, not %s", connectHandoffGraceEnv, v)
+	}
+	return d, nil
+}

internal/commands/connect_handoff_grace_dev_test.go

@@ -0,0 +1,39 @@
+//go:build dev
+
+package commands
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestConnectHandoffGraceParsesAGoDuration(t *testing.T) {
+	for in, want := range map[string]time.Duration{
+		"":      0,
+		"2s":    2 * time.Second,
+		"750ms": 750 * time.Millisecond,
+		"1m30s": 90 * time.Second,
+	} {
+		got, err := parseConnectHandoffGrace(in)
+		require.NoError(t, err, "%q", in)
+		assert.Equal(t, want, got, "%q", in)
+	}
+}
+
+func TestConnectHandoffGraceRefusesWhatIsNotAPositiveDuration(t *testing.T) {
+	for _, in := range []string{"2", "soon", "0s", "-1s"} {
+		_, err := parseConnectHandoffGrace(in)
+		require.Error(t, err, "%q", in)
+		assert.Contains(t, err.Error(), connectHandoffGraceEnv, "%q", in)
+	}
+}
+
+func TestConnectHandoffGraceReadsTheEnvironment(t *testing.T) {
+	t.Setenv(connectHandoffGraceEnv, "3s")
+	got, err := connectHandoffGrace()
+	require.NoError(t, err)
+	assert.Equal(t, 3*time.Second, got)
+}

internal/commands/connect_handoff_grace_stub.go

@@ -0,0 +1,10 @@
+//go:build !dev
+
+package commands
+
+import "time"
+
+// connectHandoffGrace is always zero in a release build, so the handoff's
+// own default applies: only a dev build lets the environment change it
+// (connect_handoff_grace_dev.go).
+func connectHandoffGrace() (time.Duration, error) { return 0, nil }

internal/commands/connect_run.go

@@ -138,6 +138,11 @@ func runConnect(cmd *cobra.Command, f *connectRunFlags) error {
 	if err != nil {
 		return err
 	}
+	// Only a dev build reads it; a release build's is always zero.
+	grace, err := connectHandoffGrace()
+	if err != nil {
+		return output.ErrUsage(err.Error())
+	}
 
 	path, err := setup.Path(config.GlobalConfigDir(), name)
 	if err != nil {
@@ -184,6 +189,9 @@ func runConnect(cmd *cobra.Command, f *connectRunFlags) error {
 	}
 
 	logger := slog.New(slog.NewTextHandler(cmd.ErrOrStderr(), nil))
+	if grace > 0 {
+		logger.Warn("connector: the handoff grace is overridden for this dev build", "grace", grace, "default", connector.HandoffGrace)
+	}
 	tokens := &managerTokens{mgr: app.Auth}
 	// Every read below needs a token first. One the running feed would
 	// wait for is waited for here too, rather than ending the start.
@@ -358,7 +366,7 @@ func runConnect(cmd *cobra.Command, f *connectRunFlags) error {
 		runPart("handoff", func(ctx context.Context) error {
 			return connector.RunHandoff(ctx, connector.HandoffOptions{
 				Ledger: ledger, Served: served.Current, Buckets: buckets, AgentID: agentID,
-				Lines: lines, Logger: logger, Started: started,
+				Lines: lines, Logger: logger, Started: started, Grace: grace,
 			})
 		})
 	}

internal/connector/handoff.go

@@ -85,6 +85,9 @@ type HandoffOptions struct {
 	// Interval is how often admitted records are looked for. Zero means
 	// DefaultHandoffInterval.
 	Interval time.Duration
+	// Grace is how long before Started a request may have been made and
+	// still be handed off. Zero means HandoffGrace.
+	Grace time.Duration
 }
 
 func (o HandoffOptions) log() *slog.Logger {
@@ -94,6 +97,13 @@ func (o HandoffOptions) log() *slog.Logger {
 	return o.Logger
 }
 
+func (o HandoffOptions) grace() time.Duration {
+	if o.Grace <= 0 {
+		return HandoffGrace
+	}
+	return o.Grace
+}
+
 // DefaultHandoffInterval is how often the handoff looks for admitted records.
 const DefaultHandoffInterval = 250 * time.Millisecond
 
@@ -164,9 +174,14 @@ func handOffReady(ctx context.Context, opts HandoffOptions) error {
 // handOff writes one record and closes it. A record made before the run
 // started is closed without being written.
 func handOff(ctx context.Context, opts HandoffOptions, record Record) error {
-	if record.CreatedAt.Before(opts.Started.Add(-HandoffGrace)) {
+	if record.CreatedAt.Before(opts.Started.Add(-opts.grace())) {
+		// Said once it is so: a run stopped before the record is closed
+		// leaves it to the next run, and must not have said it was closed.
+		if err := opts.Ledger.SetState(ctx, record.ID, StateDiscarded, ReasonBeforeThisRun); err != nil {
+			return err
+		}
 		opts.log().Info("connector: a request from before this run started is not handed off", "event_id", record.ID, "created_at", record.CreatedAt)
-		return opts.Ledger.SetState(ctx, record.ID, StateDiscarded, ReasonBeforeThisRun)
+		return nil
 	}
 	line, err := handoffLine(record, opts.AgentID)
 	if err != nil {

internal/connector/handoff_test.go

@@ -93,6 +93,51 @@ func TestARequestFromBeforeTheRunIsNotHandedOff(t *testing.T) {
 	assert.Equal(t, ReasonBeforeThisRun, r.Reason)
 }
 
+// Grace replaces HandoffGrace when set: a request made thirty seconds
+// before the run started is handed off under a grace of thirty seconds, the
+// boundary included, and not under one of twenty-nine.
+func TestTheGraceSetIsTheOneTheHandoffUses(t *testing.T) {
+	for _, tc := range []struct {
+		grace  time.Duration
+		reason string
+		lines  int
+	}{
+		{grace: 30 * time.Second, reason: ReasonHandedOff, lines: 1},
+		{grace: 29 * time.Second, reason: ReasonBeforeThisRun, lines: 0},
+	} {
+		t.Run(tc.grace.String(), func(t *testing.T) {
+			ledger := newTestLedger(t)
+			admitOn(t, ledger, 1, "recording:1")
+			var out bytes.Buffer
+			opts := handoffOptions(ledger, &out)
+			opts.Grace = tc.grace
+
+			require.NoError(t, handOffReady(context.Background(), opts))
+
+			assert.Len(t, handedOffLines(t, &out), tc.lines)
+			r := getRecord(t, ledger, 1)
+			assert.Equal(t, StateDiscarded, r.State)
+			assert.Equal(t, tc.reason, r.Reason)
+		})
+	}
+}
+
+// With no Grace set, HandoffGrace applies: a request made exactly that long
+// before the run started is still handed off.
+func TestNoGraceSetMeansHandoffGrace(t *testing.T) {
+	ledger := newTestLedger(t)
+	admitOn(t, ledger, 1, "recording:1")
+	var out bytes.Buffer
+	opts := handoffOptions(ledger, &out)
+	// testEvent's requests were made at 10:00.
+	opts.Started = time.Date(2026, 9, 16, 10, 0, 0, 0, time.UTC).Add(HandoffGrace)
+
+	require.NoError(t, handOffReady(context.Background(), opts))
+
+	assert.Len(t, handedOffLines(t, &out), 1)
+	assert.Equal(t, ReasonHandedOff, getRecord(t, ledger, 1).Reason)
+}
+
 // Only the projects the agent serves, narrowed to the run's --project.
 func TestOnlyServedProjectsAreHandedOff(t *testing.T) {
 	ledger := newTestLedger(t)