Remove what the connector's worker side left behind

internal/commands/connect_operator_test.go

@@ -160,7 +160,7 @@ func TestConnectDiscardAndRelease(t *testing.T) {
 	l, err = connector.OpenLedgerReadOnly(context.Background(), filepath.Join(dir, connector.LedgerFile))
 	require.NoError(t, err)
 	defer func() { _ = l.Close() }()
-	held, err := l.Held(ctx)
+	_, held, err := l.HoldMarker(ctx)
 	require.NoError(t, err)
 	assert.False(t, held)
 	two, _, err := l.Get(ctx, 2)

internal/commands/connect_setup_guided.go

@@ -20,7 +20,7 @@ import (
 	"github.com/basecamp/basecamp-cli/internal/auth"
 	"github.com/basecamp/basecamp-cli/internal/config"
 	"github.com/basecamp/basecamp-cli/internal/connector"
-	"github.com/basecamp/basecamp-cli/internal/connector/driver"
+	"github.com/basecamp/basecamp-cli/internal/connector/procid"
 	"github.com/basecamp/basecamp-cli/internal/connector/setup"
 	"github.com/basecamp/basecamp-cli/internal/hostutil"
 	"github.com/basecamp/basecamp-cli/internal/output"
@@ -565,7 +565,7 @@ func holderStillRuns(holder connector.InstanceHolderInfo) bool {
 	if err != nil {
 		return false
 	}
-	p, err := driver.LookupProcess(holder.PID)
+	p, err := procid.LookupProcess(holder.PID)
 	if err != nil {
 		return false
 	}

internal/commands/connect_start_wait_test.go

@@ -25,7 +25,6 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
-	"github.com/basecamp/basecamp-cli/internal/appctx"
 	"github.com/basecamp/basecamp-cli/internal/auth"
 	"github.com/basecamp/basecamp-cli/internal/config"
 	"github.com/basecamp/basecamp-cli/internal/connector"
@@ -483,55 +482,6 @@ func TestAnOversizedServerFaultIsWaitedOut(t *testing.T) {
 	assert.Equal(t, 2, mints())
 }
 
-// A second connector for an agent that is already running says so at once,
-// even while the token endpoint is rate-limiting it: the lock is checked
-// before the start waits for a token, not after.
-func TestASecondConnectorIsRefusedBeforeItWaitsForAToken(t *testing.T) {
-	f := newOperatorFixture(t)
-	dir, err := connectStateDir(f.file, false)
-	require.NoError(t, err)
-	lock, err := connector.AcquireInstanceLock(dir, f.file.AccountID, f.file.Agent.PersonID, time.Now())
-	require.NoError(t, err)
-	t.Cleanup(func() { _ = lock.Release() })
-
-	app := newConnectSetupApp(t, f.s, "agent")
-	creds, err := app.Auth.GetStore().Load(app.Auth.CredentialKey())
-	require.NoError(t, err)
-	creds.ExpiresAt = time.Now().Add(-time.Minute).Unix()
-	require.NoError(t, app.Auth.GetStore().Save(app.Auth.CredentialKey(), creds))
-	f.s.mu.Lock()
-	f.s.rateLimitMints, f.s.mints = true, 0
-	f.s.mu.Unlock()
-
-	ctx, cancel := context.WithCancel(t.Context())
-	defer cancel()
-	var buf bytes.Buffer
-	app.Output = output.New(output.Options{Format: output.FormatJSON, Writer: &buf})
-	cmd := NewConnectCmd()
-	cmd.SetArgs(nil)
-	cmd.SetContext(appctx.WithApp(ctx, app))
-	cmd.SetOut(&buf)
-	cmd.SetErr(&buf)
-	cmd.SilenceErrors, cmd.SilenceUsage = true, true
-	done := make(chan error, 1)
-	go func() { done <- cmd.Execute() }()
-
-	select {
-	case err := <-done:
-		refusal := usageError(t, err)
-		assert.Equal(t, output.CodeLockUnavailable, refusal.Code, buf.String())
-		assert.Contains(t, refusal.Message, fmt.Sprintf("held by pid %d", os.Getpid()), "the refusal names the holder")
-		assert.Contains(t, refusal.Hint, "One connector serves every repo")
-	case <-time.After(10 * time.Second):
-		cancel()
-		<-done
-		t.Fatalf("a second connector waited for a token instead of refusing: %s", buf.String())
-	}
-	f.s.mu.Lock()
-	defer f.s.mu.Unlock()
-	assert.Zero(t, f.s.mints, "nothing was minted for a connector that cannot run")
-}
-
 // A 503 that names its Retry-After is waited out for that long, by the
 // start and the feed alike, not on backoff.
 func TestAServerFaultsRetryAfterIsHonored(t *testing.T) {
@@ -567,50 +517,6 @@ func (b *lockedBuffer) String() string {
 	return b.buf.String()
 }
 
-// A start waiting for its token does not hold the agent's lock: until the
-// credential is proven to be that agent, the lock is only looked at, so a
-// profile whose credential is some other agent's cannot keep the real
-// connector out while it waits.
-func TestAStartWaitingForItsTokenHoldsNoLock(t *testing.T) {
-	f := newOperatorFixture(t)
-	app := newConnectSetupApp(t, f.s, "agent")
-	creds, err := app.Auth.GetStore().Load(app.Auth.CredentialKey())
-	require.NoError(t, err)
-	creds.ExpiresAt = time.Now().Add(-time.Minute).Unix()
-	require.NoError(t, app.Auth.GetStore().Save(app.Auth.CredentialKey(), creds))
-	f.s.mu.Lock()
-	f.s.rateLimitMints = true
-	f.s.mu.Unlock()
-
-	ctx, cancel := context.WithCancel(t.Context())
-	defer cancel()
-	var out lockedBuffer
-	app.Output = output.New(output.Options{Format: output.FormatJSON, Writer: &out})
-	cmd := NewConnectCmd()
-	cmd.SetArgs(nil)
-	cmd.SetContext(appctx.WithApp(ctx, app))
-	cmd.SetOut(&out)
-	cmd.SetErr(&out)
-	cmd.SilenceErrors, cmd.SilenceUsage = true, true
-	done := make(chan error, 1)
-	go func() { done <- cmd.Execute() }()
-
-	require.Eventually(t, func() bool { return strings.Contains(out.String(), "could not get the agent's token yet") },
-		10*time.Second, 20*time.Millisecond, "the start never waited: %s", out.String())
-	dir, err := connectStateDir(f.file, false)
-	require.NoError(t, err)
-	lock, err := connector.AcquireInstanceLock(dir, f.file.AccountID, f.file.Agent.PersonID, time.Now())
-	require.NoError(t, err, "a waiting start held the agent's lock")
-	require.NoError(t, lock.Release())
-
-	cancel()
-	select {
-	case <-done:
-	case <-time.After(10 * time.Second):
-		t.Fatal("the waiting start did not stop when canceled")
-	}
-}
-
 // okTokens is a renewal that succeeds at once.
 type okTokens struct{}
 

internal/commands/connect_start_wait_unix_test.go

@@ -0,0 +1,113 @@
+//go:build unix
+
+package commands
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"os"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/basecamp/basecamp-cli/internal/appctx"
+	"github.com/basecamp/basecamp-cli/internal/connector"
+	"github.com/basecamp/basecamp-cli/internal/output"
+)
+
+// A second connector for an agent that is already running says so at once,
+// even while the token endpoint is rate-limiting it: the lock is checked
+// before the start waits for a token, not after.
+func TestASecondConnectorIsRefusedBeforeItWaitsForAToken(t *testing.T) {
+	f := newOperatorFixture(t)
+	dir, err := connectStateDir(f.file, false)
+	require.NoError(t, err)
+	lock, err := connector.AcquireInstanceLock(dir, f.file.AccountID, f.file.Agent.PersonID, time.Now())
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = lock.Release() })
+
+	app := newConnectSetupApp(t, f.s, "agent")
+	creds, err := app.Auth.GetStore().Load(app.Auth.CredentialKey())
+	require.NoError(t, err)
+	creds.ExpiresAt = time.Now().Add(-time.Minute).Unix()
+	require.NoError(t, app.Auth.GetStore().Save(app.Auth.CredentialKey(), creds))
+	f.s.mu.Lock()
+	f.s.rateLimitMints, f.s.mints = true, 0
+	f.s.mu.Unlock()
+
+	ctx, cancel := context.WithCancel(t.Context())
+	defer cancel()
+	var buf bytes.Buffer
+	app.Output = output.New(output.Options{Format: output.FormatJSON, Writer: &buf})
+	cmd := NewConnectCmd()
+	cmd.SetArgs(nil)
+	cmd.SetContext(appctx.WithApp(ctx, app))
+	cmd.SetOut(&buf)
+	cmd.SetErr(&buf)
+	cmd.SilenceErrors, cmd.SilenceUsage = true, true
+	done := make(chan error, 1)
+	go func() { done <- cmd.Execute() }()
+
+	select {
+	case err := <-done:
+		refusal := usageError(t, err)
+		assert.Equal(t, output.CodeLockUnavailable, refusal.Code, buf.String())
+		assert.Contains(t, refusal.Message, fmt.Sprintf("held by pid %d", os.Getpid()), "the refusal names the holder")
+		assert.Contains(t, refusal.Hint, "One connector serves every repo")
+	case <-time.After(10 * time.Second):
+		cancel()
+		<-done
+		t.Fatalf("a second connector waited for a token instead of refusing: %s", buf.String())
+	}
+	f.s.mu.Lock()
+	defer f.s.mu.Unlock()
+	assert.Zero(t, f.s.mints, "nothing was minted for a connector that cannot run")
+}
+
+// A start waiting for its token does not hold the agent's lock: until the
+// credential is proven to be that agent, the lock is only looked at, so a
+// profile whose credential is some other agent's cannot keep the real
+// connector out while it waits.
+func TestAStartWaitingForItsTokenHoldsNoLock(t *testing.T) {
+	f := newOperatorFixture(t)
+	app := newConnectSetupApp(t, f.s, "agent")
+	creds, err := app.Auth.GetStore().Load(app.Auth.CredentialKey())
+	require.NoError(t, err)
+	creds.ExpiresAt = time.Now().Add(-time.Minute).Unix()
+	require.NoError(t, app.Auth.GetStore().Save(app.Auth.CredentialKey(), creds))
+	f.s.mu.Lock()
+	f.s.rateLimitMints = true
+	f.s.mu.Unlock()
+
+	ctx, cancel := context.WithCancel(t.Context())
+	defer cancel()
+	var out lockedBuffer
+	app.Output = output.New(output.Options{Format: output.FormatJSON, Writer: &out})
+	cmd := NewConnectCmd()
+	cmd.SetArgs(nil)
+	cmd.SetContext(appctx.WithApp(ctx, app))
+	cmd.SetOut(&out)
+	cmd.SetErr(&out)
+	cmd.SilenceErrors, cmd.SilenceUsage = true, true
+	done := make(chan error, 1)
+	go func() { done <- cmd.Execute() }()
+
+	require.Eventually(t, func() bool { return strings.Contains(out.String(), "could not get the agent's token yet") },
+		10*time.Second, 20*time.Millisecond, "the start never waited: %s", out.String())
+	dir, err := connectStateDir(f.file, false)
+	require.NoError(t, err)
+	lock, err := connector.AcquireInstanceLock(dir, f.file.AccountID, f.file.Agent.PersonID, time.Now())
+	require.NoError(t, err, "a waiting start held the agent's lock")
+	require.NoError(t, lock.Release())
+
+	cancel()
+	select {
+	case <-done:
+	case <-time.After(10 * time.Second):
+		t.Fatal("the waiting start did not stop when canceled")
+	}
+}

internal/commands/connect_worker.go

@@ -6,13 +6,13 @@ import (
 	"errors"
 
 	"github.com/basecamp/basecamp-cli/internal/connector"
-	"github.com/basecamp/basecamp-cli/internal/connector/driver"
+	"github.com/basecamp/basecamp-cli/internal/connector/procid"
 )
 
-// The operator commands act on a recorded worker only through the driver's
-// one-owner rule (driver/worker.go): a pid is not an identity, so every
-// question about a recorded worker is driver.OwnsWorker's, and nothing here
-// tests a pid of its own.
+// Status asks about a worker an older build recorded only through the
+// one-owner rule (procid): a pid is not an identity, so every question about
+// a recorded worker is procid.OwnsWorker's, and nothing here tests a pid of
+// its own.
 
 // Worker states the operator commands report.
 const (
@@ -22,7 +22,7 @@ const (
 	workerUnverified  = "unverified"
 	workerNotRecorded = "not_recorded"
 	// workerUnaccounted is the task token's holder alone: it was delivered
-	// and the connector cannot name who has it (connector.TokenHolder).
+	// and the connector that delivered it could not name who has it.
 	workerUnaccounted = "unaccounted"
 )
 
@@ -39,9 +39,9 @@ func recordedWorkerState(t connector.TaskStatus) string {
 // A holder the connector could not account for is answered before the kernel
 // is asked anything: there is no pid to ask about, and "not recorded" would
 // read as nothing having taken the token, which is the opposite of what
-// happened. It is the same distinction the release point acts on
-// (connector.TokenHolder), and this is where the person who must settle the
-// held attempt reads it.
+// happened. The connector that delivered it held the attempt on that
+// distinction, and this is where the person who must settle the held
+// attempt reads it.
 func recordedTakerState(t connector.TaskStatus) string {
 	if t.TakerUnaccounted {
 		return workerUnaccounted
@@ -52,15 +52,15 @@ func recordedTakerState(t connector.TaskStatus) string {
 // recordedProcessState asks the one-owner rule about a process the ledger
 // recorded. It takes the identity the ledger builds (TaskStatus's own
 // WorkerIdentity and TakerIdentity) rather than assembling one from parts:
-// a driver.Process put together here would have no StartedExact, and the
+// a procid.Process put together here would have no StartedExact, and the
 // rule answers ErrIdentityUnknown to that — every live worker would read as
 // "unverified".
-func recordedProcessState(p driver.Process) string {
+func recordedProcessState(p procid.Process) string {
 	if p.PID <= 0 || p.PGID <= 0 || p.StartedAt.IsZero() {
 		return workerNotRecorded
 	}
-	switch owns, err := driver.OwnsWorker(p); {
-	case errors.Is(err, driver.ErrGroupOutlivedLeader):
+	switch owns, err := procid.OwnsWorker(p); {
+	case errors.Is(err, procid.ErrGroupOutlivedLeader):
 		return workerHeld
 	case err != nil:
 		return workerUnverified

internal/commands/connect_worker_other.go

@@ -9,7 +9,7 @@ import (
 // The one-owner rule is Unix's: process groups, and a start time that makes a
 // pid an identity, are what it rests on. Elsewhere the connector does not run
 // (the run command refuses), and nothing here can say whether a recorded
-// worker is still itself — so nothing is signaled and nothing is claimed.
+// worker is still itself, so nothing is claimed.
 
 const (
 	workerUnverified  = "unverified"

internal/commands/connect_worker_unix_test.go

@@ -13,9 +13,9 @@ import (
 // A token delivered to a process the connector could not name is the one
 // case the zero taker cannot express, and status is where the person who
 // must settle the held attempt reads it. "Nothing took the token" and "the
-// token is out and nobody can say who has it" are opposite facts; the same
-// conflation the dispatcher stopped making (TokenHolder.Unaccounted) must
-// not survive on the operator's side of it.
+// token is out and nobody can say who has it" are opposite facts, and the
+// removed dispatcher held the attempt on the difference; it must not be lost
+// on the operator's side of it.
 func TestStatusSaysWhenATokenHolderCannotBeAccountedFor(t *testing.T) {
 	assert.Equal(t, workerUnaccounted,
 		recordedTakerState(connector.TaskStatus{TakerUnaccounted: true}),

internal/connector/blocked_retry_test.go

@@ -507,30 +507,6 @@ func TestAStaleRetryHandoffIsNotDecidedInsideTheInterval(t *testing.T) {
 	assert.True(t, ok, "and at the moment the schedule names, it loads")
 }
 
-// A person's redispatch is not a timer, and does not wait for one. It marks
-// the record due, so the rerun it asks for runs at once — and so a rerun that
-// never happened (the command died between the authorization and the
-// admission run) is picked up by the next sweep rather than stranded.
-func TestARedispatchMakesABlockedRecordDueAtOnce(t *testing.T) {
-	ledger, clock := retryLedger(t)
-	ctx := context.Background()
-	blockRecord(t, ledger, 1, adapterBucketID, admission.ReasonReadFailed)
-
-	_, ok, err := ledger.Admission().LoadUndecided(ctx, 1)
-	require.NoError(t, err)
-	require.False(t, ok, "nine minutes early, on the timer's account")
-
-	out, err := ledger.Redispatch(ctx, 1, "jorge", []int64{adapterBucketID})
-	require.NoError(t, err)
-	require.True(t, out.Rerun)
-
-	_, ok, err = ledger.Admission().LoadUndecided(ctx, 1)
-	require.NoError(t, err)
-	assert.True(t, ok, "a person asked for it now")
-	assert.Equal(t, []int64{1}, dueIDs(t, ledger, BlockedRetryScope{Now: clock.Now(), Limit: 10}),
-		"and if the rerun never happens, the sweep picks it up")
-}
-
 // Copilot on #770, and it is the starvation the card named, reintroduced
 // through the back door. The claim stops a row being offered twice; it does
 // nothing if the claimed rows still spend the sweep's budget. A limit filled
@@ -573,15 +549,8 @@ func TestABlockedRecordWithNoScheduledAttemptIsNotRunnable(t *testing.T) {
 	_, ok, err := ledger.Admission().LoadUndecided(ctx, 1)
 	require.NoError(t, err)
 	assert.False(t, ok, "nothing has given this record an attempt to make")
-
-	// A person is what gives an untimed record one.
-	_, err = ledger.Redispatch(ctx, 1, "jorge", []int64{adapterBucketID})
-	require.NoError(t, err)
-	_, ok, err = ledger.Admission().LoadUndecided(ctx, 1)
-	require.NoError(t, err)
-	assert.True(t, ok, "the redispatch is the attempt")
-	assert.Equal(t, []int64{1}, dueIDs(t, ledger, BlockedRetryScope{Now: clock.Now(), Limit: 10}),
-		"and if the rerun never happens, the sweep picks it up rather than stranding it")
+	assert.Empty(t, dueIDs(t, ledger, BlockedRetryScope{Now: clock.Now(), Limit: 10}),
+		"and the sweep does not offer it either")
 }
 
 // queryPlan is what SQLite says it will actually do, which is the only

internal/connector/doc.go

@@ -7,7 +7,7 @@
 // Feed rows are pointers — id, type, bucket, creator, recording — and nothing
 // else. No title, no body, no URL, no names. Intake writes that pointer to the
 // ledger if the id is new and hands the id to a queue. That is the whole of
-// the work on the feed's delivery path: a slow admission or a busy dispatcher
+// the work on the feed's delivery path: a slow admission or a slow hand-off
 // is absorbed by the backlog rather than felt by the socket.
 //
 // Absorbed, not unbounded. The backlog is a bounded queue, and at its pause