From 4d1cf96f9d02f53c32e079274626d316e0544448 Mon Sep 17 00:00:00 2001 From: Jorge Manrubia Date: Sat, 19 Sep 2026 09:20:32 +0200 Subject: [PATCH 1/4] The served set is held, not just read, from check to commit MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A launch and a `connect setup --unserve` could interleave: the dispatcher read connect.json once per pass and handed that reading down to LaunchTask, and `connect redispatch` read the file at start-up and passed those buckets to Ledger.Redispatch with the ledger opened in between. An unserve landing in either window still started one task. The exposure was one — the next tick re-reads and stops the follow-ups — and it was pre-existing in kind rather than a regression, but the served set is the whole boundary now, so the window is worth closing. Both sites now read the served set under `.connect.lock`, the per-profile lock `connect setup` already holds across its load, change and save, and keep the lock until the ledger has committed. The reading is fresh under the lock: the TTL cache stays what admission and each dispatcher pass read, and reusing it here would reintroduce the gap the lock is taken to close. Where the lock goes and what happens when it cannot be taken: - connectServed.Authorize takes the lock, reloads, and hands back the release. Launch and redispatch use it; nothing else does. Admission keeps Current and its TTL — it decides once per event, and a file lock per event is not a thing to put on that path. - The dispatcher never waits: setup.TryLock, and a pass that loses gives up its turn for one tick. Waiting would park the hot path behind a `connect setup`'s network checks. Nothing is lost; the record is still waiting next pass. - `connect redispatch` is an operator command, so it waits, bounded, and reports busy and retryable when a setup outlasts the wait. - setup.Lock now waits up to ten seconds rather than refusing on sight, so a dispatcher pass that overlaps a setup cannot fail `connect setup` for a reason nobody could act on. - A lock nobody could take is kept apart from a lock somebody holds: ErrPolicyUnreadable and ErrPolicyBusy. Both authorize nothing; only one means something is wrong. What is held under the lock is one file read and one SQLite transaction, released before a session directory, a token socket or a worker process exists. The TTL is now asserted as well as documented (the second card). The window it bounds does not close: admission still reads through the cache, so a record can be admitted into a project unserved within the TTL, which dispatch then refuses and the stranded report names. Two tests pin it — the constant at two seconds, and that the cache is real, so the constant is the bound rather than a number nothing obeys. Co-Authored-By: Claude Opus 5 (1M context) --- internal/commands/connect_operator.go | 65 ++++- internal/commands/connect_policy_lock_test.go | 236 ++++++++++++++++++ internal/commands/connect_run.go | 89 ++++++- .../connector/dispatch_policy_lock_test.go | 147 +++++++++++ internal/connector/dispatcher.go | 142 +++++++++-- internal/connector/dispatcher_test.go | 40 ++- internal/connector/ledger_decisions.go | 7 +- internal/connector/ledger_tasks.go | 23 +- .../connector/operator_invariants_test.go | 5 +- internal/connector/recovery_connector_test.go | 5 +- internal/connector/setup/private.go | 68 ++++- internal/connector/setup/private_unix_test.go | 55 +++- skills/basecamp-connect/SKILL.md | 4 +- 13 files changed, 821 insertions(+), 65 deletions(-) create mode 100644 internal/commands/connect_policy_lock_test.go create mode 100644 internal/connector/dispatch_policy_lock_test.go diff --git a/internal/commands/connect_operator.go b/internal/commands/connect_operator.go index f906f3065..e6df0c532 100644 --- a/internal/commands/connect_operator.go +++ b/internal/commands/connect_operator.go @@ -35,14 +35,12 @@ import ( type connectProfile struct { app *appctx.App name string + path string file setup.File } // servedBucketsOf is the projects a connect.json serves, as the ledger's -// decisions want them, from the file this command loaded when it started. -// That is a reading and not a lock: nothing stops `connect setup --unserve` -// completing between it and the write below, and holding the setup lock -// across both is carded rather than done here. +// decisions want them. func servedBucketsOf(file setup.File) []int64 { served := make([]int64, 0, len(file.Projects)) for bucket := range file.Projects { @@ -52,6 +50,42 @@ func servedBucketsOf(file setup.File) []int64 { return served } +// servedBucketsUnderLock is the served projects read under connect.json's own +// lock, with the release to hold until the decision that reading authorizes +// has been written. The file this command loaded when it started is not that +// reading: a `connect setup --unserve` can complete between start-up and the +// write, and the point here is that it cannot complete between this read and +// the release. +// +// It waits for a holder rather than refusing on sight (setup.Lock, not +// TryLock): this is an operator's command, a connector's own hold is +// measured in milliseconds, and a person who typed a redispatch would rather +// wait out a `connect setup` than be told to run it again. A holder that +// outlasts the wait is reported as busy and retryable, and nothing is +// written. +// +// The lock is taken before the ledger is written and released after, never +// the other way about, so the two locks are always taken in one order. +func servedBucketsUnderLock(p connectProfile) ([]int64, func(), error) { + unlock, err := setup.Lock(p.path) + if err != nil { + return nil, nil, classifyLockError(p.name, err) + } + file, err := setup.Load(p.path) + if err != nil { + unlock() + return nil, nil, output.ErrUsage("connect.json cannot be used: " + setup.ErrorText(err)) + } + // The same refusal the connector's own reader makes: a file that now + // names another agent or account is not this profile's policy, and a + // decision taken against it would be taken for somebody else. + if file.Agent != p.file.Agent || file.AccountID != p.file.AccountID { + unlock() + return nil, nil, output.ErrUsage("connect.json now names another agent or account, so nothing was decided") + } + return servedBucketsOf(file), unlock, nil +} + func loadConnectProfile(cmd *cobra.Command) (connectProfile, error) { app := appctx.FromContext(cmd.Context()) if app == nil { @@ -75,7 +109,7 @@ func loadConnectProfile(cmd *cobra.Command) (connectProfile, error) { case err != nil: return connectProfile{}, output.ErrUsage("connect.json cannot be used: " + err.Error()) } - return connectProfile{app: app, name: name, file: file}, nil + return connectProfile{app: app, name: name, path: path, file: file}, nil } // operatorName is who a decision is recorded as: the local user who ran it. @@ -446,12 +480,21 @@ func runConnectRedispatch(cmd *cobra.Command, raw string) error { } defer done() - // The served projects as this command read them, rather than the bit on - // the record: a record admitted while its project was served is not - // authorization to run it after the operator stopped serving it. Read at - // start-up and not re-read here, so an unserve landing in between is not - // caught — see servedBucketsOf. - res, err := ledger.Redispatch(ctx, id, operatorName(), servedBucketsOf(p.file)) + // The served projects as connect.json has them now, rather than the bit + // on the record: a record admitted while its project was served is not + // authorization to run it after the operator stopped serving it. Read + // under the setup lock and the lock held until the ledger has written, + // so an unserve lands wholly before the reading or wholly after the + // decision — not between them. + served, release, err := servedBucketsUnderLock(p) + if err != nil { + return err + } + res, err := ledger.Redispatch(ctx, id, operatorName(), served) + // Let go before the prerequisite is re-run below: that talks to + // Basecamp, and nothing that waits on the network is held under a lock + // `connect setup` waits on. + release() if err != nil { return decisionError(err) } diff --git a/internal/commands/connect_policy_lock_test.go b/internal/commands/connect_policy_lock_test.go new file mode 100644 index 000000000..829a8081a --- /dev/null +++ b/internal/commands/connect_policy_lock_test.go @@ -0,0 +1,236 @@ +package commands + +import ( + "encoding/json" + "log/slog" + "os" + "path/filepath" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/basecamp/basecamp-cli/internal/config" + "github.com/basecamp/basecamp-cli/internal/connector" + "github.com/basecamp/basecamp-cli/internal/connector/admission" + "github.com/basecamp/basecamp-cli/internal/connector/setup" + "github.com/basecamp/basecamp-cli/internal/output" +) + +// policyFile is a connect.json on disk where setup and the connector both +// expect one: under the profile's own directory, owner-only, with the +// directories above it owner-only too. Nothing here fakes the lock or the +// read — the tests below take the real flock on the real file. +func policyFile(t *testing.T, projects map[int64]admission.Project) (string, setup.File) { + t.Helper() + t.Setenv("XDG_CONFIG_HOME", t.TempDir()) + path, err := setup.Path(config.GlobalConfigDir(), "agent") + require.NoError(t, err) + require.NoError(t, os.MkdirAll(filepath.Dir(path), 0o700)) + + file := setup.New("agent") + file.AccountID = "2914079" + file.Agent = setup.Agent{PersonID: 52007412, Kind: setup.KindAgent} + file.Trust.OperatorID = 26909558 + file.Projects = projects + writePolicyFile(t, path, file) + return path, file +} + +func writePolicyFile(t *testing.T, path string, file setup.File) { + t.Helper() + require.NoError(t, file.Validate()) + data, err := json.Marshal(file) + require.NoError(t, err) + require.NoError(t, os.WriteFile(path, data, 0o600)) +} + +func servingNothing(file setup.File) setup.File { + out := file + out.Projects = map[int64]admission.Project{} + return out +} + +// Authorize reads connect.json under the lock rather than reusing what the +// TTL cache has: the clock does not move here, so a cached answer would still +// be inside its TTL and would still name the project the operator has just +// unserved. That reused reading is the whole of the window this closes. +func TestAuthorizeReadsTheFileRatherThanTheCache(t *testing.T) { + path, file := policyFile(t, map[int64]admission.Project{48929974: {Class: "internal"}}) + clock := time.Date(2026, 9, 19, 12, 0, 0, 0, time.UTC) + served := newConnectServed(path, file, slog.New(slog.DiscardHandler)) + served.now = func() time.Time { return clock } + + warm, err := served.Current() + require.NoError(t, err) + require.Contains(t, warm, int64(48929974), "the cache holds a reading that serves the project") + + writePolicyFile(t, path, servingNothing(file)) + + projects, release, err := served.Authorize() + require.NoError(t, err) + t.Cleanup(release) + assert.Empty(t, projects, "the launch is authorized against the file, not against a reading taken before the unserve") + + // And the cache is not left older than what was just read, so the next + // caller cannot go back to the answer this one refused. + current, err := served.Current() + require.NoError(t, err) + assert.Empty(t, current) +} + +// The lock is really taken and really let go: while Authorize's release is +// held nobody else can lock the profile's policy, and once it is released +// they can. A `connect setup --unserve` is on the far side of exactly this. +func TestAuthorizeHoldsThePolicyLockUntilItIsReleased(t *testing.T) { + path, file := policyFile(t, map[int64]admission.Project{48929974: {}}) + served := newConnectServed(path, file, slog.New(slog.DiscardHandler)) + + _, release, err := served.Authorize() + require.NoError(t, err) + + _, err = setup.TryLock(path) + assert.ErrorIs(t, err, setup.ErrSetupRunning, "a setup cannot change the file while a launch is being authorized against it") + + release() + unlock, err := setup.TryLock(path) + require.NoError(t, err, "and it can the moment the launch has committed") + unlock() +} + +// The other direction: a policy somebody else holds authorizes nothing, and +// says so as ErrPolicyBusy rather than as an empty served set — the +// dispatcher gives up its pass, and nothing tells an operator their projects +// stopped being served. +func TestAuthorizeIsBusyWhileSomebodyElseHoldsThePolicy(t *testing.T) { + path, file := policyFile(t, map[int64]admission.Project{48929974: {}}) + served := newConnectServed(path, file, slog.New(slog.DiscardHandler)) + + unlock, err := setup.TryLock(path) + require.NoError(t, err) + t.Cleanup(unlock) + + start := time.Now() + projects, release, err := served.Authorize() + assert.ErrorIs(t, err, connector.ErrPolicyBusy) + assert.NotErrorIs(t, err, connector.ErrPolicyUnreadable, "somebody holding the lock is not the same as nothing being able to read it") + assert.Nil(t, projects, "a lock nobody could take is not an empty served set") + assert.Nil(t, release) + assert.Less(t, time.Since(start), setup.LockWait/2, "the dispatcher's read does not wait on a file another process writes") +} + +// A connect.json that no longer loads is the other kind of nothing: it +// authorizes none either, and says which it is rather than reporting an +// empty served set or a busy lock. +func TestAuthorizeReportsAPolicyThatCannotBeReadAsUnreadable(t *testing.T) { + path, file := policyFile(t, map[int64]admission.Project{48929974: {}}) + served := newConnectServed(path, file, slog.New(slog.DiscardHandler)) + require.NoError(t, os.WriteFile(path, []byte("{not json"), 0o600)) + + projects, release, err := served.Authorize() + assert.ErrorIs(t, err, connector.ErrPolicyUnreadable) + assert.NotErrorIs(t, err, connector.ErrPolicyBusy) + assert.Nil(t, projects) + assert.Nil(t, release) + + unlock, err := setup.TryLock(path) + require.NoError(t, err, "and the lock it took is not left behind") + unlock() +} + +// connectServedTTL is the width of the one disagreement the lock does not +// close: admission decides against a cached reading, dispatch against a +// locked one, so a project unserved within the TTL can still be admitted — +// and is then refused at the launch and named by the stranded report. +// Raising this constant widens that, and a bound only a comment knows is a +// description rather than a guarantee. +func TestConnectServedTTLStaysSmall(t *testing.T) { + assert.Positive(t, connectServedTTL, "a reading reused for no time at all reads connect.json once per event") + assert.LessOrEqual(t, connectServedTTL, 2*time.Second, + "connectServedTTL is how stale admission's view of the served set may be; raising it is a decision, not an edit") +} + +// And the constant is the bound because the cache is real: a reading inside +// the TTL is reused, so the number is what an admission decision can be +// behind the file. Asserting the constant without this would pin a number +// nothing obeys. +func TestAReadingIsReusedWithinItsTTL(t *testing.T) { + path, file := policyFile(t, map[int64]admission.Project{48929974: {}}) + clock := time.Date(2026, 9, 19, 12, 0, 0, 0, time.UTC) + served := newConnectServed(path, file, slog.New(slog.DiscardHandler)) + served.now = func() time.Time { return clock } + + first, err := served.Current() + require.NoError(t, err) + require.Contains(t, first, int64(48929974)) + + writePolicyFile(t, path, servingNothing(file)) + clock = clock.Add(connectServedTTL - time.Nanosecond) + within, err := served.Current() + require.NoError(t, err) + assert.Contains(t, within, int64(48929974), "inside the TTL the reading is reused, which is what the TTL bounds") + + clock = clock.Add(time.Nanosecond) + after, err := served.Current() + require.NoError(t, err) + assert.Empty(t, after, "and at the TTL it is read again") +} + +// `connect redispatch` authorizes against connect.json as it is when it +// writes, not as it was when the command started. The profile it loaded at +// start-up still serves the project; the file no longer does. +func TestRedispatchReadsTheServedSetUnderTheLock(t *testing.T) { + path, file := policyFile(t, map[int64]admission.Project{48929974: {}, 48699913: {}}) + p := connectProfile{name: "agent", path: path, file: file} + + stale := servingNothing(file) + stale.Projects = map[int64]admission.Project{48929974: {}} + writePolicyFile(t, path, stale) + + served, release, err := servedBucketsUnderLock(p) + require.NoError(t, err) + t.Cleanup(release) + assert.Equal(t, []int64{48929974}, served, "the unserved project is gone from the set the decision is written against") + + _, err = setup.TryLock(path) + assert.ErrorIs(t, err, setup.ErrSetupRunning, "and the lock is held until the decision has been written") +} + +// A setup that outlasts the wait leaves the redispatch unwritten, reported +// as busy and retryable rather than as a policy that refuses the project. +func TestRedispatchIsBusyWhileSetupHoldsThePolicy(t *testing.T) { + path, file := policyFile(t, map[int64]admission.Project{48929974: {}}) + was := setup.LockWait + setup.LockWait = 20 * time.Millisecond + t.Cleanup(func() { setup.LockWait = was }) + + unlock, err := setup.TryLock(path) + require.NoError(t, err) + t.Cleanup(unlock) + + _, _, err = servedBucketsUnderLock(connectProfile{name: "agent", path: path, file: file}) + require.Error(t, err) + var apiErr *output.Error + require.ErrorAs(t, err, &apiErr) + assert.Equal(t, output.CodeBusy, apiErr.Code) + assert.True(t, apiErr.Retryable, "machine consumers read retryable, not the code") +} + +// A file that has been pointed at another agent since the command started is +// not this profile's policy, and a decision taken against it would be taken +// for somebody else. +func TestRedispatchRefusesAPolicyThatNowNamesAnotherAgent(t *testing.T) { + path, file := policyFile(t, map[int64]admission.Project{48929974: {}}) + other := file + other.Agent.PersonID = 1 + writePolicyFile(t, path, other) + + _, _, err := servedBucketsUnderLock(connectProfile{name: "agent", path: path, file: file}) + require.Error(t, err) + assert.Contains(t, err.Error(), "another agent or account") + + unlock, err := setup.TryLock(path) + require.NoError(t, err, "a refusal leaves no lock behind") + unlock() +} diff --git a/internal/commands/connect_run.go b/internal/commands/connect_run.go index d70878c8b..cc1bdabf5 100644 --- a/internal/commands/connect_run.go +++ b/internal/commands/connect_run.go @@ -330,9 +330,17 @@ func runConnect(cmd *cobra.Command, f *connectRunFlags) error { // gap is seen by one and not the other (Copilot on #765). The // difference from the bug this replaced is that the disagreement is // bounded: one decision against a served set at most connectServedTTL - // old, where the startup file never caught up at all. Holding one - // snapshot across a whole decision needs a lock over setup, which is - // carded rather than done here. + // old, where the startup file never caught up at all. That bound is + // asserted, not merely described — see TestConnectServedTTLStaysSmall. + // + // One place does hold a snapshot across a whole decision, and it is the + // one where a stale reading would start work: served.Authorize reads + // connect.json under `connect setup`'s own lock and keeps the lock until + // the launch has committed. Admission keeps the cache: it decides once + // per event, a file lock per event is not a thing to put on that path, + // and the cost of it being a reading rather than a lock is a record + // admitted into a project that has just stopped being served — which + // dispatch then refuses and reportStranded names. served := newConnectServed(path, file, logger) reads := admission.NewSDKReads(&basecamp.Config{BaseURL: app.Config.BaseURL}, tokens, account, connectSDKOptions()...) @@ -372,7 +380,8 @@ func runConnect(cmd *cobra.Command, f *connectRunFlags) error { return output.ErrUsage(err.Error()) } options := connectDispatcherOptions(connectDispatch{ - File: file, Buckets: buckets, Ledger: ledger, Driver: worker, Served: served.Current, + File: file, Buckets: buckets, Ledger: ledger, Driver: worker, + Served: served.Current, Authorize: served.Authorize, Profile: name, Executable: exe, StateDir: stateDir, SessionsDir: sessions, // Replies are listed with their words, so the connector's own // notices are left out even before their receipts are known, and @@ -576,6 +585,9 @@ type connectServed struct { // safe to tell a person on a card (Copilot on #765). err error failing bool + // lockFailing is failing's counterpart for the lock: a host that cannot + // take it says so once rather than once per launch. + lockFailing bool } // connectServedTTL is how long a read of connect.json is reused. @@ -587,11 +599,14 @@ func newConnectServed(path string, file setup.File, log *slog.Logger) *connectSe // Current returns a copy of the projects connect.json serves as of the last // read, or the reason it could not be read. "As of the last read" is the -// honest span: a reading is reused for connectServedTTL, and nothing holds -// the setup lock, so a `connect setup --unserve` can complete between any -// read and whatever the caller goes on to do with it. Dispatch treats an error as authorizing +// honest span: a reading is reused for connectServedTTL, and no lock is +// taken, so a `connect setup --unserve` can complete between any read and +// whatever the caller goes on to do with it. Dispatch treats an error as authorizing // nothing; admission holds the record as a configuration error rather than // answering that the project is not served. +// +// A caller that must not have an unserve land between its reading and what +// that reading authorizes wants Authorize, not this. func (r *connectServed) Current() (map[int64]admission.Project, error) { r.mu.Lock() defer r.mu.Unlock() @@ -602,6 +617,62 @@ func (r *connectServed) Current() (map[int64]admission.Project, error) { if (r.projects == nil && r.err == nil) || r.now().Sub(r.loadedAt) >= connectServedTTL { r.reload() } + return r.answer() +} + +// Authorize reads connect.json under the per-profile setup lock — the lock +// `connect setup` holds across its load, change and save — and hands back +// the release for the caller to hold until the commit that reading +// authorizes has landed. Between the two an unserve cannot complete, so the +// launch is decided against the file as it will still be when the task +// exists. +// +// The read is fresh: the TTL is a reuse policy for callers that are only +// choosing, and reusing a reading here would reintroduce exactly the gap the +// lock is taken to close. It refreshes the cache on its way through, so the +// next Current cannot be older than this read. +// +// A lock another command holds is connector.ErrPolicyBusy: this authorizes +// nothing, and the dispatcher gives up its pass rather than waiting out a +// `connect setup`'s network checks. A host that cannot lock at all +// authorizes nothing either — the guarantee here is the lock, as it is in +// setup. +func (r *connectServed) Authorize() (map[int64]admission.Project, func(), error) { + unlock, err := setup.TryLock(r.path) + if err != nil { + if errors.Is(err, setup.ErrSetupRunning) { + return nil, nil, fmt.Errorf("%w: %w", connector.ErrPolicyBusy, err) + } + r.logLockFailure(err) + return nil, nil, fmt.Errorf("%w: %w", connector.ErrPolicyUnreadable, err) + } + r.mu.Lock() + r.lockFailing = false + r.reload() + projects, err := r.answer() + r.mu.Unlock() + if err != nil { + unlock() + // reload has already said why, once. This only tells the dispatcher + // which kind of nothing it is being handed. + return nil, nil, fmt.Errorf("%w: %w", connector.ErrPolicyUnreadable, err) + } + return projects, unlock, nil +} + +// logLockFailure says a lock could not be taken, once per spell of failing. +func (r *connectServed) logLockFailure(err error) { + r.mu.Lock() + defer r.mu.Unlock() + if !r.lockFailing { + r.log.Error("connector: dispatching nothing until connect.json can be locked", "error", err) + } + r.lockFailing = true +} + +// answer is the reader's answer from whatever the last reload left: a copy of +// the served projects, or the reason there is none. r.mu is held. +func (r *connectServed) answer() (map[int64]admission.Project, error) { if r.err != nil { return nil, r.err } @@ -647,6 +718,9 @@ type connectDispatch struct { Ledger *connector.Ledger Driver driver.Driver Served func() (map[int64]admission.Project, error) + // Authorize is Served read under connect.json's lock, with the release + // the dispatcher holds across a launch's commit. + Authorize func() (map[int64]admission.Project, func(), error) Profile string Executable string @@ -667,6 +741,7 @@ func connectDispatcherOptions(d connectDispatch) connector.DispatcherOptions { Ledger: d.Ledger, Driver: d.Driver, Served: d.Served, + Authorize: d.Authorize, Concurrency: d.File.Concurrency, Deadline: time.Duration(d.File.Deadline), Buckets: d.Buckets, diff --git a/internal/connector/dispatch_policy_lock_test.go b/internal/connector/dispatch_policy_lock_test.go new file mode 100644 index 000000000..d6d090789 --- /dev/null +++ b/internal/connector/dispatch_policy_lock_test.go @@ -0,0 +1,147 @@ +package connector + +import ( + "errors" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/basecamp/basecamp-cli/internal/connector/admission" + "github.com/basecamp/basecamp-cli/internal/connector/driver" +) + +// The served set the launch is decided against is read under connect.json's +// lock and the lock is held until the launch has committed, so an unserve +// lands wholly before the reading or wholly after the task exists. Here it +// lands between the pass's reading and the commit — the window this closes — +// and the task the old order would have started is not started. +// +// The pass still reads the project as served: that reading chooses the +// records worth trying, and choosing a record the launch then refuses is the +// shape this wants, not a disagreement. +func TestALaunchIsDecidedAgainstThePolicyAtTheCommit(t *testing.T) { + fake := newFakeDriver() + h := newDispatchHarness(t, fake, nil) + h.authorized = map[int64]admission.Project{} + admitOn(t, h.ledger, 1, "recording:1") + h.run(t) + time.Sleep(150 * time.Millisecond) + + fake.mu.Lock() + assert.Empty(t, fake.sessions, "an unserve that lands before the commit stops the task it would have started") + fake.mu.Unlock() + assert.Equal(t, StateAdmitted, getRecord(t, h.ledger, 1).State, "and the record is left where it was") +} + +// A pass that cannot take the policy lock authorizes nothing and loses +// nothing. `connect setup` holds that lock across its network checks, so the +// dispatcher gives up its turn rather than blocking on it; the record is +// still waiting on the next tick. +func TestAPassAuthorizesNothingWhileThePolicyIsBusy(t *testing.T) { + fake := newFakeDriver() + h := newDispatchHarness(t, fake, nil) + h.authorizeErr = ErrPolicyBusy + admitOn(t, h.ledger, 1, "recording:1") + h.run(t) + time.Sleep(150 * time.Millisecond) + + fake.mu.Lock() + require.Empty(t, fake.sessions, "nothing is started while somebody else is changing connect.json") + fake.mu.Unlock() + require.Equal(t, StateAdmitted, getRecord(t, h.ledger, 1).State) + + h.mu.Lock() + h.authorizeErr = nil + h.mu.Unlock() + select { + case <-fake.made: + case <-time.After(5 * time.Second): + t.Fatal("the record was not started once the policy was free again") + } +} + +// A reading that cannot be taken at all — a host that cannot lock, a +// connect.json that no longer loads — authorizes nothing either, and is kept +// apart from busy: one says somebody is changing the file, the other says +// something is wrong. Failing closed is the same posture the reader takes, +// and setup takes, when the answer cannot be read. +func TestAPassAuthorizesNothingWhenThePolicyCannotBeRead(t *testing.T) { + for name, cause := range map[string]error{ + "the policy could not be read under its lock": ErrPolicyUnreadable, + "something nothing here classified": errors.New("connect.json cannot be locked"), + } { + t.Run(name, func(t *testing.T) { + fake := newFakeDriver() + h := newDispatchHarness(t, fake, nil) + h.authorizeErr = cause + admitOn(t, h.ledger, 1, "recording:1") + h.run(t) + time.Sleep(150 * time.Millisecond) + + fake.mu.Lock() + assert.Empty(t, fake.sessions) + fake.mu.Unlock() + assert.Equal(t, StateAdmitted, getRecord(t, h.ledger, 1).State) + }) + } +} + +// What is held under the policy lock is one file read and one ledger +// transaction: it is let go when the launch has committed, before a session +// directory, a token socket or a worker process exists. A dispatcher that +// held it across a spawn would park `connect setup` behind a worker. +func TestThePolicyLockIsLetGoBeforeAWorkerStarts(t *testing.T) { + fake := newFakeDriver() + var h *dispatchHarness + held := make(chan bool, 4) + fake.onStart = func(driver.SessionConfig) { + h.mu.Lock() + defer h.mu.Unlock() + held <- h.policyHeld + } + h = newDispatchHarness(t, fake, nil) + admitOn(t, h.ledger, 1, "recording:1") + h.run(t) + + select { + case stillHeld := <-held: + assert.False(t, stillHeld, "the policy lock is released when the ledger has committed, not when the worker has started") + case <-time.After(5 * time.Second): + t.Fatal("no worker started") + } +} + +// The dispatcher needs both readings, and says which is missing: one to +// choose with, one to authorize with. A dispatcher built without the second +// would launch against whatever the first happened to say. +func TestADispatcherNeedsThePolicyUnderItsLock(t *testing.T) { + fake := newFakeDriver() + h := newDispatchHarness(t, fake, nil) + opts := h.d.opts + opts.Authorize = nil + _, err := NewDispatcher(opts) + require.Error(t, err) + assert.Contains(t, err.Error(), "under its lock") +} + +// The launch defers Authorize's release, so a nil one would be a panic on the +// dispatcher's own path rather than a policy it could not read. Nothing in +// the tree returns one; the guard is for whatever wires this next. +func TestALaunchSurvivesAnAuthorizeThatReturnsNoRelease(t *testing.T) { + fake := newFakeDriver() + h := newDispatchHarness(t, fake, func(o *DispatcherOptions) { + o.Authorize = func() (map[int64]admission.Project, func(), error) { + return map[int64]admission.Project{adapterBucketID: {}}, nil, nil + } + }) + admitOn(t, h.ledger, 1, "recording:1") + h.run(t) + + select { + case <-fake.made: + case <-time.After(5 * time.Second): + t.Fatal("no worker started") + } +} diff --git a/internal/connector/dispatcher.go b/internal/connector/dispatcher.go index 004b9dbd4..073ef9c8e 100644 --- a/internal/connector/dispatcher.go +++ b/internal/connector/dispatcher.go @@ -66,6 +66,24 @@ const ( // tools are mcp__basecamp__*. const MCPServerName = "basecamp" +// ErrPolicyBusy is Authorize's answer when another process holds +// connect.json's lock — a `connect setup` mid-run, which holds it across its +// network checks. It authorizes nothing and means nothing is wrong: the pass +// dispatches none and asks again on the next tick, and so does every pass +// until that setup lets go. A dispatcher that waited here instead would park +// its hot path on a file another process writes, which is a worse thing to +// own than a queue that drains the moment the policy is settled. +var ErrPolicyBusy = errors.New("connector: connect.json is being changed by another command") + +// ErrPolicyUnreadable is Authorize's answer when the served set could not be +// taken under its lock at all: a host that cannot lock, a connect.json that +// no longer loads. It authorizes nothing, as ErrPolicyBusy does, and unlike +// ErrPolicyBusy something is wrong — kept apart from it for the same reason +// a failed read is kept apart from an empty served set. Every pass +// dispatches nothing until it is fixed; the reader is what says so, once, +// rather than once per tick. +var ErrPolicyUnreadable = errors.New("connector: connect.json could not be read under its lock") + // ReplyLister lists the agent's comments or chat lines at a reply destination, // for the adopted-reply rule. type ReplyLister interface { @@ -87,6 +105,27 @@ type DispatcherOptions struct { // is that nothing could read the file, is the same false claim the // holding reply used to make on a card (Copilot on #765). Served func() (map[int64]admission.Project, error) + // Authorize is the same served set read under connect.json's own lock — + // the one `connect setup` holds across its load, change and save — with + // the release the caller holds across the commit that reading + // authorizes. Nothing is cached: the read happens under the lock, so an + // unserve cannot complete between it and the launch it permits. + // + // It is not Served with a lock bolted on, and it does not replace + // Served. Served is what a pass reads to choose from and to keep a + // worker's follow-ups flowing — cheap, cached, called per event by + // admission, and a file lock per event is exactly the shape that would + // not survive contact with the hot path. Authorize is taken once per + // launch, held across one SQLite transaction, and released before a + // worker is spawned. + // + // The release is never nil when the error is nil. A lock another + // process holds is reported as ErrPolicyBusy and authorizes nothing: + // the pass dispatches none and tries again on the next tick, rather + // than blocking behind a `connect setup`'s network checks. A lock or a + // file that could not be read at all is ErrPolicyUnreadable, which + // authorizes nothing either and says that something is wrong. + Authorize func() (map[int64]admission.Project, func(), error) // TokenWindow is how long a task token's socket waits for the worker's // MCP server; DefaultTokenWindow when zero. TokenWindow time.Duration @@ -207,6 +246,8 @@ func NewDispatcher(opts DispatcherOptions) (*Dispatcher, error) { return nil, errors.New("connector: the dispatcher needs a driver") case opts.Served == nil: return nil, errors.New("connector: the dispatcher needs connect.json's served projects") + case opts.Authorize == nil: + return nil, errors.New("connector: the dispatcher needs connect.json's served projects under its lock") case opts.MCP.Command == "" || opts.MCP.Profile == "" || opts.MCP.StateDir == "": return nil, errors.New("connector: the dispatcher needs the worker's MCP server command, profile and state directory") case opts.PrivateDir == "": @@ -448,7 +489,8 @@ func (d *Dispatcher) dispatchReady(ctx context.Context) error { } // Invariant 2, in the query: only records in a project this pass read as // served, among the projects this run hears. A record the dispatcher - // cannot start never fills the window. + // cannot start never fills the window. It narrows the candidates; the + // launch decides, against its own locked reading. records, err := d.ledger.StartableRecordsWhere(ctx, StartableFilter{ Served: served, Limit: d.opts.Concurrency * 4, }) @@ -463,10 +505,30 @@ func (d *Dispatcher) dispatchReady(ctx context.Context) error { if d.free() <= 0 { break } - if err := d.start(ctx, record, served); err != nil { - if errors.Is(err, ErrNotStartable) { - continue - } + switch err := d.start(ctx, record); { + case err == nil: + case errors.Is(err, ErrNotStartable): + // Including a record whose project the launch's own reading no + // longer serves: the unserve won, and the next pass will not + // offer it again. + continue + case errors.Is(err, ErrPolicyBusy): + // Somebody is changing connect.json. Nothing is wrong and + // nothing is lost: this pass authorizes nothing and the next + // tick asks again. Every pass for as long as that setup runs + // gives up the same way, so the connector starts nothing while + // its policy is being rewritten — which is the right thing to + // do with a policy somebody is in the middle of changing, and + // is cheaper than a dispatcher blocked on a file `connect + // setup` holds across its network checks. + d.log.Debug("connector: connect.json is being changed; nothing is dispatched this pass") + return nil + case errors.Is(err, ErrPolicyUnreadable): + // Something is wrong, and the reader has already said so once. + // Saying it again on every tick would bury it; dispatching + // nothing is the whole of what this pass can do about it. + return nil + default: return err } } @@ -524,6 +586,12 @@ func (d *Dispatcher) reportStranded(ctx context.Context, served []int64, servedE // or one reused within its TTL, never a lock — narrowed to the ones this run // hears. One pass takes it once and hands it down, so everything that pass // decides is decided from the same reading. +// +// It chooses; it does not authorize. What a pass reads here picks the records +// worth trying and keeps a live worker's follow-ups flowing. The launch +// itself re-reads under the lock (authorizedBuckets), so a record this +// reading offered can still be refused a moment later, and that is the point +// rather than a disagreement. func (d *Dispatcher) servedBuckets() ([]int64, error) { projects, err := d.opts.Served() if err != nil { @@ -532,6 +600,36 @@ func (d *Dispatcher) servedBuckets() ([]int64, error) { // which is why the empty set and the error travel together. return nil, err } + return d.narrow(projects), nil +} + +// authorizedBuckets is the served projects read under connect.json's lock, +// narrowed to the ones this run hears, with the release to hold across the +// commit that reading authorizes. A `connect setup --unserve` cannot complete +// between the read and the release, so the launch under it is decided against +// the file as it is at the commit, not as it was when the pass began. +// +// The release is never nil when the error is nil, and the caller lets go of +// it as soon as the ledger has committed: nothing that talks to a worker, +// a driver or the network happens under this lock. +func (d *Dispatcher) authorizedBuckets() ([]int64, func(), error) { + projects, release, err := d.opts.Authorize() + // Never nil either way, whatever Authorize returned: the caller defers + // the release, and a nil one would turn a policy that could not be read + // into a panic on the dispatcher's own path. + if release == nil { + release = func() {} + } + if err != nil { + release() + return nil, nil, err + } + return d.narrow(projects), release, nil +} + +// narrow keeps the served projects this run hears: --project scope, or every +// served project when the run named none. +func (d *Dispatcher) narrow(projects map[int64]admission.Project) []int64 { var served []int64 for bucket := range projects { if len(d.opts.Buckets) == 0 || slices.Contains(d.opts.Buckets, bucket) { @@ -539,24 +637,23 @@ func (d *Dispatcher) servedBuckets() ([]int64, error) { } } slices.Sort(served) - return served, nil + return served } // start launches a task for record: the ledger first, then the driver, and // the release point on every path that fails after it. Capacity is the // caller's question (free), not this one's. // -// served is the snapshot the record was chosen against, handed down rather -// than read again: one pass of the dispatcher decides from one reading of -// connect.json, as one verdict does (admission's policyNow). -func (d *Dispatcher) start(ctx context.Context, record Record, served []int64) error { +// The served set the launch is decided against is read here rather than +// handed down from the pass, and it is read under connect.json's own lock, +// held until the launch transaction has committed. That is what makes an +// unserve and a launch one order rather than a race: the file the launch +// reads is the file as it will still be when the task exists. +func (d *Dispatcher) start(ctx context.Context, record Record) error { // Nothing is prepared and nothing is resolved: the worker runs where the // connector was started, and a task that needs a clone or a directory of // its own is the agent's business to make. - launch, err := d.ledger.LaunchTask(ctx, LaunchSpec{ - EventID: record.ID, Served: served, - Driver: d.opts.Driver.Name(), Deadline: d.opts.Deadline, - }) + launch, err := d.launch(ctx, record) if err != nil { return err } @@ -629,6 +726,23 @@ func (d *Dispatcher) start(ctx context.Context, record Record, served []int64) e return nil } +// launch commits the ledger's side of a start, under connect.json's lock and +// against a reading taken under it. The lock is let go the moment the +// transaction returns — before a session directory, a token socket or a +// worker process exists — so what is held across it is one file read and one +// SQLite transaction, and nothing that waits on a person or the network. +func (d *Dispatcher) launch(ctx context.Context, record Record) (Launch, error) { + served, release, err := d.authorizedBuckets() + if err != nil { + return Launch{}, err + } + defer release() + return d.ledger.LaunchTask(ctx, LaunchSpec{ + EventID: record.ID, Served: served, + Driver: d.opts.Driver.Name(), Deadline: d.opts.Deadline, + }) +} + // sessionConfig builds what the driver is given (invariant 3). func (d *Dispatcher) sessionConfig(ctx context.Context, launch Launch, record Record) (driver.SessionConfig, *TokenSocket, func(), error) { dir := filepath.Join(d.opts.PrivateDir, launch.AttemptID) diff --git a/internal/connector/dispatcher_test.go b/internal/connector/dispatcher_test.go index 4749e0126..1ded93144 100644 --- a/internal/connector/dispatcher_test.go +++ b/internal/connector/dispatcher_test.go @@ -167,7 +167,18 @@ type dispatchHarness struct { served map[int64]admission.Project // servedErr stands in for a connect.json that cannot be read. servedErr error - mu sync.Mutex + // authorized, when non-nil, is what Authorize answers instead of served: + // the launch's own reading of connect.json under its lock, which a test + // makes differ from the reading the pass chose with to stand for a + // `connect setup --unserve` landing between the two. + authorized map[int64]admission.Project + // authorizeErr stands in for a policy lock this process could not take — + // a `connect setup` holding it, or a host that cannot lock at all. + authorizeErr error + // policyHeld is true while the dispatcher holds what Authorize handed + // it, so a test can see where the lock is let go. + policyHeld bool + mu sync.Mutex } func newDispatchHarness(t *testing.T, fake *fakeDriver, tweak func(*DispatcherOptions)) *dispatchHarness { @@ -194,6 +205,33 @@ func newDispatchHarness(t *testing.T, fake *fakeDriver, tweak func(*DispatcherOp } return out, nil }, + // The launch's own reading, under the lock. It answers from served + // unless a test has set authorized, so a test that does not care + // about the difference sees one connect.json. + Authorize: func() (map[int64]admission.Project, func(), error) { + h.mu.Lock() + defer h.mu.Unlock() + if h.authorizeErr != nil { + return nil, nil, h.authorizeErr + } + source, err := h.authorized, error(nil) + if source == nil { + source, err = h.served, h.servedErr + } + if err != nil { + return nil, nil, err + } + out := map[int64]admission.Project{} + for k, v := range source { + out[k] = v + } + h.policyHeld = true + return out, func() { + h.mu.Lock() + defer h.mu.Unlock() + h.policyHeld = false + }, nil + }, WorkDir: testWorkDir, Concurrency: 2, Deadline: time.Hour, diff --git a/internal/connector/ledger_decisions.go b/internal/connector/ledger_decisions.go index d008ca56d..121c7a8db 100644 --- a/internal/connector/ledger_decisions.go +++ b/internal/connector/ledger_decisions.go @@ -190,8 +190,11 @@ func (l *Ledger) redispatch(ctx context.Context, eventID int64, by string, serve // reported success and left the record for a dispatcher that will refuse // to launch it is worse than one that refuses here (Copilot on #765). // - // served is a reading, not a lock: `connect setup --unserve` landing - // between that read and this commit is not caught here. Carded. + // served is a reading its caller took under connect.json's own lock and + // holds until this commits (`connect redispatch`'s + // servedBucketsUnderLock), so an unserve lands wholly before that read + // or wholly after this write. This checks the set it was given; the + // caller is what makes the set current. dispatchable := !record.ContentDropped && len(record.Decision.Snapshot) > 0 && record.Decision.Served && slices.Contains(served, record.BucketID) && record.Decision.ConversationKey != "" diff --git a/internal/connector/ledger_tasks.go b/internal/connector/ledger_tasks.go index 73264aee1..78b853684 100644 --- a/internal/connector/ledger_tasks.go +++ b/internal/connector/ledger_tasks.go @@ -324,18 +324,19 @@ func (l *Ledger) launchTask(ctx context.Context, spec LaunchSpec, attemptID stri // spec.Served is the set its caller decided this launch from. // // What this buys, said plainly, because the comment here used to - // claim more. For the dispatcher it is redundant with the startable - // query, which filtered on the same slice — belt and braces at the - // ledger's own boundary, so a launch that names a set not covering - // its record is refused rather than trusted, and one that names no - // set authorizes nothing rather than defaulting open. + // claim more. It is a check on the set its caller named, not a read + // of the file: a launch that names a set not covering its record is + // refused rather than trusted, and one that names no set authorizes + // nothing rather than defaulting open. That is what makes a + // non-dispatcher caller fail closed. // - // What it does NOT do is close the window between reading - // connect.json and committing this transaction. An unserve landing - // in that window still launches this one task; a follow-up is - // stopped at the next tick by taskRun.authorized. That race is - // pre-existing in kind — the route had it too — and holding the - // setup lock across selection and commit is carded, not done here. + // What closes the window between reading connect.json and + // committing this transaction is the caller's, and it is now held: + // Dispatcher.launch and `connect redispatch` read the served set + // under connect.json's own lock and keep the lock until this + // commits, so an unserve lands wholly before the reading or wholly + // after the task exists. This check is what that reading is spent + // on. return Launch{}, fmt.Errorf("connector: launch event %d in project %d, which is not served: %w", spec.EventID, record.BucketID, ErrNotStartable) } var busy bool diff --git a/internal/connector/operator_invariants_test.go b/internal/connector/operator_invariants_test.go index c1d01e6d8..00b0fa487 100644 --- a/internal/connector/operator_invariants_test.go +++ b/internal/connector/operator_invariants_test.go @@ -66,8 +66,9 @@ func unknownOutcome(t *testing.T, l *Ledger, id int64) Launch { // Copilot on #765: a redispatch is authorized by the served projects its // caller read from connect.json, not by the bit admission wrote when it -// decided the record. A reading, not a lock — an unserve landing between -// that read and this write is not caught, and is carded. +// decided the record. The caller reads that set under connect.json's lock +// and holds the lock until this write, so an unserve cannot land between +// the two (TestRedispatchReadsTheServedSetUnderTheLock, in commands). // // Without this the command reported success and admitted the record, and the // dispatcher then refused to launch it — so the person was told their diff --git a/internal/connector/recovery_connector_test.go b/internal/connector/recovery_connector_test.go index f96883b2e..3a67f32d5 100644 --- a/internal/connector/recovery_connector_test.go +++ b/internal/connector/recovery_connector_test.go @@ -300,7 +300,10 @@ func runHarnessConnector(dir string) error { worker := &failingSpawns{Driver: working, broken: d.New(filepath.Join(dir, "no-such-agent")), failures: failures} dispatcher, err := NewDispatcher(DispatcherOptions{ Ledger: ledger, Driver: worker, - Served: func() (map[int64]admission.Project, error) { return served, nil }, + Served: func() (map[int64]admission.Project, error) { return served, nil }, + Authorize: func() (map[int64]admission.Project, func(), error) { + return served, func() {}, nil + }, WorkDir: work, Concurrency: 2, Deadline: time.Hour, diff --git a/internal/connector/setup/private.go b/internal/connector/setup/private.go index 05c722eee..f07a5f3d7 100644 --- a/internal/connector/setup/private.go +++ b/internal/connector/setup/private.go @@ -6,6 +6,7 @@ import ( "io" "os" "path/filepath" + "time" "github.com/gofrs/flock" ) @@ -50,24 +51,59 @@ func ensurePrivateDirs(path string) error { return nil } -// ErrSetupRunning reports another setup already running for this profile. -var ErrSetupRunning = errors.New("another connect setup is running for this profile") +// ErrSetupRunning reports the per-profile policy lock held by somebody else. +// Two holders take it: `connect setup`, across its whole load-change-save, +// and a running connector, for the moment it authorizes a launch or a +// redispatch against the file. +var ErrSetupRunning = errors.New("another command holds this profile's connector policy") // ErrLockUnavailable reports a host that cannot take the setup lock at all: // a filesystem without flock, a lock file this user does not own. Setup's // guarantee is the lock, so it refuses rather than run without one. var ErrLockUnavailable = errors.New("this host cannot lock the connector's policy") -// Lock takes the per-profile setup lock beside connect.json, so two setups -// cannot interleave a load, a change and a save. It refuses rather than -// waits: a second setup on one profile is a mistake to report, not a queue. +// LockWait bounds Lock's wait for a holder to finish. The connector takes +// this lock too now, and what it does under it is one file read and one +// SQLite transaction — bounded by the ledger's own five-second busy retry — +// so a setup that refused the instant a dispatcher pass overlapped it would +// fail for no reason a person could act on. Ten seconds is that worst case +// with room, and a setup that waits it out reports a real holder. +// +// A variable so tests can shorten it. +var LockWait = 10 * time.Second + +// lockPoll is how often a waiter re-tries. flock has no blocking-with-timeout +// form through this library, so the wait is a poll — the same shape, and for +// the same reason, as the credential lock's (internal/auth/lock.go). +const lockPoll = 20 * time.Millisecond + +// Lock takes the per-profile setup lock beside connect.json, so a load, a +// change and a save cannot interleave with another holder's. It waits up to +// LockWait and then reports ErrSetupRunning: the wait is for the connector's +// brief hold, not a queue of setups, and a second setup on one profile is +// still a mistake to report rather than to serialize — it will not have +// finished inside LockWait. func Lock(path string) (unlock func(), err error) { + return lock(path, LockWait) +} + +// TryLock takes the lock if it is free and reports ErrSetupRunning if it is +// not, without waiting. It is the dispatcher's form: a pass that cannot take +// the lock authorizes nothing and tries again on its next tick, which costs +// one tick of latency. Waiting instead would park the dispatcher behind a +// `connect setup`'s network checks, and a dispatcher that blocks on a file +// another process writes is its own hazard. +func TryLock(path string) (unlock func(), err error) { + return lock(path, 0) +} + +func lock(path string, wait time.Duration) (unlock func(), err error) { if err := ensurePrivateDirs(path); err != nil { return nil, err } lockPath := filepath.Join(filepath.Dir(path), ".connect.lock") // The lock is only a lock if it is this user's own file: a symlink or a - // foreign file left in the directory could point two setups at + // foreign file left in the directory could point two holders at // different inodes, and they would not exclude each other. if err := checkPrivateLockFile(lockPath); err != nil { if errors.Is(err, ErrNotPrivate) { @@ -78,14 +114,20 @@ func Lock(path string) (unlock func(), err error) { return nil, fmt.Errorf("%w: %s: %w", ErrLockUnavailable, lockPath, err) } lock := flock.New(lockPath, flock.SetPermissions(0o600)) - held, err := lock.TryLock() - if err != nil { - return nil, fmt.Errorf("%w: %s: %w", ErrLockUnavailable, lockPath, err) - } - if !held { - return nil, fmt.Errorf("%w: %s", ErrSetupRunning, filepath.Dir(path)) + deadline := time.Now().Add(wait) + for { + held, err := lock.TryLock() + if err != nil { + return nil, fmt.Errorf("%w: %s: %w", ErrLockUnavailable, lockPath, err) + } + if held { + return func() { _ = lock.Unlock() }, nil + } + if !time.Now().Before(deadline) { + return nil, fmt.Errorf("%w: %s", ErrSetupRunning, filepath.Dir(path)) + } + time.Sleep(lockPoll) } - return func() { _ = lock.Unlock() }, nil } // checkPrivateLockFile refuses a lock file this user does not solely own. A diff --git a/internal/connector/setup/private_unix_test.go b/internal/connector/setup/private_unix_test.go index c6a0f4bfe..328174949 100644 --- a/internal/connector/setup/private_unix_test.go +++ b/internal/connector/setup/private_unix_test.go @@ -8,6 +8,7 @@ import ( "path/filepath" "syscall" "testing" + "time" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -173,14 +174,24 @@ func TestLoadFollowsAnOwnAncestorSymlinkAndChecksItsTarget(t *testing.T) { assert.True(t, errors.Is(err, ErrNotPrivate), "the target's modes count: %v", err) } +// shortLockWait keeps a test that means to see contention from sitting out +// the whole of LockWait to see it. +func shortLockWait(t *testing.T) { + t.Helper() + was := LockWait + LockWait = 20 * time.Millisecond + t.Cleanup(func() { LockWait = was }) +} + func TestLockRefusesASecondSetup(t *testing.T) { + shortLockWait(t) path, err := Path(configDir(t), "agent") require.NoError(t, err) unlock, err := Lock(path) require.NoError(t, err) _, err = Lock(path) - assert.Error(t, err) + assert.ErrorIs(t, err, ErrSetupRunning) unlock() unlockAgain, err := Lock(path) @@ -188,6 +199,48 @@ func TestLockRefusesASecondSetup(t *testing.T) { unlockAgain() } +// The dispatcher's form: it never waits, so a pass that overlaps a setup +// gives up its turn instead of parking on a file another process writes. +func TestTryLockDoesNotWaitForAHolder(t *testing.T) { + path, err := Path(configDir(t), "agent") + require.NoError(t, err) + unlock, err := TryLock(path) + require.NoError(t, err) + + // LockWait is left long on purpose: if TryLock ever waited, this would + // take it rather than return inside the window asserted below. + start := time.Now() + _, err = TryLock(path) + assert.ErrorIs(t, err, ErrSetupRunning) + assert.Less(t, time.Since(start), LockWait/2, "TryLock returns rather than waits") + + unlock() + again, err := TryLock(path) + require.NoError(t, err) + again() +} + +// Lock waits for a holder that finishes inside LockWait, so a `connect +// setup` does not fail because a dispatcher pass happened to overlap it. +func TestLockWaitsForAHolderThatFinishes(t *testing.T) { + path, err := Path(configDir(t), "agent") + require.NoError(t, err) + unlock, err := TryLock(path) + require.NoError(t, err) + + released := make(chan struct{}) + go func() { + time.Sleep(5 * lockPoll) + unlock() + close(released) + }() + + got, err := Lock(path) + require.NoError(t, err, "the wait outlasts a holder that lets go") + <-released + got() +} + func TestCheckPrivateFileCreatesNothingAndHoldsTheRules(t *testing.T) { dir := filepath.Join(t.TempDir(), "state") require.NoError(t, os.Mkdir(dir, 0o700)) diff --git a/skills/basecamp-connect/SKILL.md b/skills/basecamp-connect/SKILL.md index 7308ffecb..986746393 100644 --- a/skills/basecamp-connect/SKILL.md +++ b/skills/basecamp-connect/SKILL.md @@ -121,7 +121,7 @@ the link and code while it waits; then wait for it to finish. |------|-------| | Credential | The CLI's credential store, under the profile. `basecamp auth status -P '' --json` describes it (see Inspecting). Never open it. | | connect.json | `$XDG_CONFIG_HOME/basecamp/connect//connect.json`, default `~/.config/basecamp/connect//connect.json`. Setup's JSON result gives the exact `path`. | -| Setup lock | `.connect.lock` beside connect.json. One setup per profile at a time. | +| Setup lock | `.connect.lock` beside connect.json. One setup per profile at a time. A running connector takes it too, for the moment it authorizes a launch or a redispatch against the file, so setup waits briefly rather than refusing when a dispatcher pass overlaps it. | | Connector runtime state (ledger, checkpoint, lock) | `$XDG_STATE_HOME/basecamp/connect/-/`, default under `~/.local/state`; a shadow run's is under `connect-shadow/` instead. Read it only through `basecamp connect status` and `basecamp connect doctor`; never open or copy the files. | The CLI's configuration, its profiles and (when it uses files) its credential @@ -371,7 +371,7 @@ the exit status: exit 7 is shared. | `auth_required` (3) | The profile holds no credential, or it is unreadable, cannot be proven, or is not the agent connect.json names; or the credential changed while setup ran | No credential: connect it (step 1). Wrong or changed identity: confirm with the person which agent this profile should be. Changed mid-run: run setup again. | | `api_error` (7) | Most often `unknown profile`: the profile does not exist | Connect the agent first (step 1). | | `not_ready` (7) | A readiness check failed. `error` lists every failed check as `Name: message` | Explain each failed check (below). | -| `busy` (5) | Another command is using this profile's credential or setup lock | Nothing is wrong. Run setup again when it has finished. | +| `busy` (5) | Another command is using this profile's credential or setup lock — another setup, or a connector authorizing a launch | Nothing is wrong. Run setup again when it has finished. | | `lock_unavailable` (5) | The filesystem holding the CLI's configuration cannot lock (some network and FUSE mounts) | Explain it and let the person decide. The fix is a local filesystem for `XDG_CONFIG_HOME`, and moving it hides every profile and stored file credential. After such a move do not reconnect the agent: that rotates its secret. | A setup the person stops also writes nothing. From 32389ca74adc8f1331f8e92a0ea869d08227588b Mon Sep 17 00:00:00 2001 From: Jorge Manrubia Date: Sat, 19 Sep 2026 09:53:48 +0200 Subject: [PATCH 2/4] Close the review comments on the served-set lock MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Five findings from Copilot on #771, and one from an adversarial pass that Copilot did not raise. The one that mattered: a blocked record's redispatch re-runs what blocked it, and `Ledger.Redispatch` sets that rerun without consulting the served set at all. The fresh locked reading only narrowed the ledger's decision; the rerun still built its policy from the file the command loaded at start-up, so an unserve landing in between could produce an admitted verdict for a project the operator had withdrawn. `servedBucketsUnderLock` is now `authorizedProfile`, which returns the whole profile with the locked reading in it, and the caller reassigns its own profile from it — the start-up reading is gone from the scope rather than merely unused. The wait honors its context. `Lock` takes one now and polls on a select, with a check before each attempt as well as around the sleep, so a person who stops the command during contention stops waiting, and a context that has already ended takes the lock from nobody rather than acquiring it and going on to act under it. What a caller that cannot take the lock gets is a refusal, never a fall-through: "the policy could not be checked" is not permission to change it. `TryLock` stays context-free, because it does not wait. Who holds this lock, and for how long, is written where the lock is defined: setup across its whole run including network checks, redispatch across one read and one ledger write, the connector once per launch and never waiting — with the rule a fourth caller needs, that the ledger's locks are taken under this one and never the other way about. `connect redispatch` no longer borrows setup's recovery text: a lock failure there names the command that actually failed, and stops claiming only setup needs the lock. Two contention tests were inheriting the production wait and spending twenty seconds proving the clock works; they shorten it now, and a duration assertion of my own came out of a test about what an answer is. Two things the adversarial pass caught and Copilot did not. Every earlier test still passed when the release was moved to before the launch transaction — the lock taken rather than held — so the release now reads the ledger and the test asserts the record had already moved. And invariant 2 claimed a worker stops being handed follow-ups "the moment" its project stops being served, which was never true and is not true now: a follow-up runs on the task's own goroutine, where a busy lock and "no more work" are the same answer, so contention there would end tasks early. The follow-up path keeps the reader's TTL deliberately, and the invariant says so. Co-Authored-By: Claude Opus 5 (1M context) --- internal/commands/connect.go | 11 ++- internal/commands/connect_operator.go | 67 ++++++++++++---- internal/commands/connect_policy_lock_test.go | 78 +++++++++++++++++-- internal/commands/connect_setup_test.go | 17 +++- .../connector/dispatch_policy_lock_test.go | 32 ++++++++ internal/connector/dispatcher.go | 32 ++++++-- internal/connector/setup/private.go | 65 +++++++++++++--- internal/connector/setup/private_unix_test.go | 54 ++++++++++++- skills/basecamp-connect/SKILL.md | 4 +- 9 files changed, 317 insertions(+), 43 deletions(-) diff --git a/internal/commands/connect.go b/internal/commands/connect.go index 8ccf841a3..a795897b9 100644 --- a/internal/commands/connect.go +++ b/internal/commands/connect.go @@ -397,7 +397,9 @@ func runConnectSetup(cmd *cobra.Command, app *appctx.App, f *connectSetupFlags) return output.ErrUsage(err.Error()) } // One setup per profile at a time: load, change and save are one step. - unlock, err := setup.Lock(path) + // The wait honors ctx, so a person who stops the command during + // contention is not made to sit out the whole of it. + unlock, err := setup.Lock(ctx, path) if err != nil { return classifyLockError(name, err) } @@ -614,7 +616,7 @@ func runConnectSetup(cmd *cobra.Command, app *appctx.App, f *connectSetupFlags) // classifyWriteError puts the last step's failures in the command's exit // contract: a credential that is gone or unreadable is auth, another -// process holding the profile's credential or setup lock is busy, a +// process holding the profile's credential or the profile's policy lock is busy, a // connect.json nobody else may change is usage. func classifyWriteError(name string, err error) error { var apiErr *output.Error @@ -653,6 +655,11 @@ func classifyWriteError(name string, err error) error { // lock_unavailable, and a connect.json nobody else may change stays usage. func classifyLockError(name string, err error) error { switch { + case errors.Is(err, context.Canceled), errors.Is(err, context.DeadlineExceeded): + // A person who stopped the command, or a deadline, while waiting for + // the lock: the run ended, nothing was locked, and nothing was + // changed. Reported as itself rather than dressed as a lock failure. + return err case errors.Is(err, setup.ErrSetupRunning): return errBusy(name, err) case errors.Is(err, setup.ErrLockUnavailable): diff --git a/internal/commands/connect_operator.go b/internal/commands/connect_operator.go index e6df0c532..c071cad7f 100644 --- a/internal/commands/connect_operator.go +++ b/internal/commands/connect_operator.go @@ -50,12 +50,20 @@ func servedBucketsOf(file setup.File) []int64 { return served } -// servedBucketsUnderLock is the served projects read under connect.json's own -// lock, with the release to hold until the decision that reading authorizes -// has been written. The file this command loaded when it started is not that -// reading: a `connect setup --unserve` can complete between start-up and the -// write, and the point here is that it cannot complete between this read and -// the release. +// authorizedProfile is p with connect.json as it is now — read under the +// file's own lock — together with the projects it serves and the release to +// hold until the decision that reading authorizes has been written. The file +// the command loaded when it started is not that reading: a `connect setup +// --unserve` can complete between start-up and the write, and the point here +// is that it cannot complete between this read and the release. +// +// It returns the whole profile and not only the bucket ids because a +// redispatch is not one decision. A blocked record's redispatch re-runs what +// blocked it, and that rerun builds an admission policy from the file; a +// rerun carrying the start-up file can re-admit a record whose project was +// unserved in between and write a verdict that is not true of the policy as +// it stands (Copilot on #771). The caller reassigns its own profile from +// this, so the start-up reading is gone rather than merely unused. // // It waits for a holder rather than refusing on sight (setup.Lock, not // TryLock): this is an operator's command, a connector's own hold is @@ -66,24 +74,53 @@ func servedBucketsOf(file setup.File) []int64 { // // The lock is taken before the ledger is written and released after, never // the other way about, so the two locks are always taken in one order. -func servedBucketsUnderLock(p connectProfile) ([]int64, func(), error) { - unlock, err := setup.Lock(p.path) +func authorizedProfile(ctx context.Context, p connectProfile) (connectProfile, []int64, func(), error) { + unlock, err := setup.Lock(ctx, p.path) if err != nil { - return nil, nil, classifyLockError(p.name, err) + return p, nil, nil, classifyDecisionLockError(p.name, err) } file, err := setup.Load(p.path) if err != nil { unlock() - return nil, nil, output.ErrUsage("connect.json cannot be used: " + setup.ErrorText(err)) + return p, nil, nil, output.ErrUsage("connect.json cannot be used: " + setup.ErrorText(err)) } // The same refusal the connector's own reader makes: a file that now // names another agent or account is not this profile's policy, and a // decision taken against it would be taken for somebody else. if file.Agent != p.file.Agent || file.AccountID != p.file.AccountID { unlock() - return nil, nil, output.ErrUsage("connect.json now names another agent or account, so nothing was decided") + return p, nil, nil, output.ErrUsage("connect.json now names another agent or account, so nothing was decided") } - return servedBucketsOf(file), unlock, nil + p.file = file + return p, servedBucketsOf(file), unlock, nil +} + +// classifyDecisionLockError is classifyLockError's wording for an operator's +// decision rather than for setup. The two say the same things about the same +// failures and name different commands to run again: telling somebody whose +// redispatch was refused to "run setup again" sends them somewhere else at +// the moment they are already puzzled, and saying only setup needs the lock +// is no longer true (Copilot on #771). +func classifyDecisionLockError(name string, err error) error { + profile := shellQuote(name) + switch { + case errors.Is(err, context.Canceled), errors.Is(err, context.DeadlineExceeded): + // Stopped while waiting for the lock: nothing was locked, nothing + // was decided, and the interruption is what to report. + return err + case errors.Is(err, setup.ErrSetupRunning): + return &output.Error{Code: output.CodeBusy, Retryable: true, + Message: fmt.Sprintf("Another command is working on profile %q right now, so nothing was decided: %s", name, setup.ErrorText(err)), + Hint: "Nothing is wrong with the profile. Run the command again when it has finished: basecamp connect redispatch -P " + profile + " "} + case errors.Is(err, setup.ErrLockUnavailable): + return &output.Error{Code: output.CodeLockUnavailable, + Message: fmt.Sprintf("This host cannot lock profile %q's connector policy: %s", name, setup.ErrorText(err)), + Hint: "Nothing was decided. A decision reads which projects are served under that lock, as setup writes them under it, so this needs a filesystem for XDG_CONFIG_HOME that supports locking " + + "(some network and FUSE mounts do not). Moving it hides every profile and stored file credential."} + case errors.Is(err, setup.ErrNotPrivate): + return output.ErrUsageHint(err.Error(), "A decision reads connect.json only where nobody else can change it.") + } + return output.ErrUsage(err.Error()) } func loadConnectProfile(cmd *cobra.Command) (connectProfile, error) { @@ -486,7 +523,11 @@ func runConnectRedispatch(cmd *cobra.Command, raw string) error { // under the setup lock and the lock held until the ledger has written, // so an unserve lands wholly before the reading or wholly after the // decision — not between them. - served, release, err := servedBucketsUnderLock(p) + // p itself is reassigned, so the reading the command started with is + // gone rather than merely unused: everything below — the ledger's + // decision and the rerun's policy alike — is the file as it was when + // this was authorized. + p, served, release, err := authorizedProfile(ctx, p) if err != nil { return err } diff --git a/internal/commands/connect_policy_lock_test.go b/internal/commands/connect_policy_lock_test.go index 829a8081a..245fb22de 100644 --- a/internal/commands/connect_policy_lock_test.go +++ b/internal/commands/connect_policy_lock_test.go @@ -1,10 +1,12 @@ package commands import ( + "context" "encoding/json" "log/slog" "os" "path/filepath" + "strings" "testing" "time" @@ -111,13 +113,14 @@ func TestAuthorizeIsBusyWhileSomebodyElseHoldsThePolicy(t *testing.T) { require.NoError(t, err) t.Cleanup(unlock) - start := time.Now() + // That this read does not wait is setup.TryLock's property and is + // asserted there (TestTryLockDoesNotWaitForAHolder), where the clock is + // the thing under test. Here the question is only what it answers. projects, release, err := served.Authorize() assert.ErrorIs(t, err, connector.ErrPolicyBusy) assert.NotErrorIs(t, err, connector.ErrPolicyUnreadable, "somebody holding the lock is not the same as nothing being able to read it") assert.Nil(t, projects, "a lock nobody could take is not an empty served set") assert.Nil(t, release) - assert.Less(t, time.Since(start), setup.LockWait/2, "the dispatcher's read does not wait on a file another process writes") } // A connect.json that no longer loads is the other kind of nothing: it @@ -188,10 +191,11 @@ func TestRedispatchReadsTheServedSetUnderTheLock(t *testing.T) { stale.Projects = map[int64]admission.Project{48929974: {}} writePolicyFile(t, path, stale) - served, release, err := servedBucketsUnderLock(p) + authorized, served, release, err := authorizedProfile(context.Background(), p) require.NoError(t, err) t.Cleanup(release) assert.Equal(t, []int64{48929974}, served, "the unserved project is gone from the set the decision is written against") + assert.NotContains(t, authorized.file.Projects, int64(48699913), "and gone from the profile everything below decides with") _, err = setup.TryLock(path) assert.ErrorIs(t, err, setup.ErrSetupRunning, "and the lock is held until the decision has been written") @@ -209,7 +213,7 @@ func TestRedispatchIsBusyWhileSetupHoldsThePolicy(t *testing.T) { require.NoError(t, err) t.Cleanup(unlock) - _, _, err = servedBucketsUnderLock(connectProfile{name: "agent", path: path, file: file}) + _, _, _, err = authorizedProfile(context.Background(), connectProfile{name: "agent", path: path, file: file}) require.Error(t, err) var apiErr *output.Error require.ErrorAs(t, err, &apiErr) @@ -226,7 +230,7 @@ func TestRedispatchRefusesAPolicyThatNowNamesAnotherAgent(t *testing.T) { other.Agent.PersonID = 1 writePolicyFile(t, path, other) - _, _, err := servedBucketsUnderLock(connectProfile{name: "agent", path: path, file: file}) + _, _, _, err := authorizedProfile(context.Background(), connectProfile{name: "agent", path: path, file: file}) require.Error(t, err) assert.Contains(t, err.Error(), "another agent or account") @@ -234,3 +238,67 @@ func TestRedispatchRefusesAPolicyThatNowNamesAnotherAgent(t *testing.T) { require.NoError(t, err, "a refusal leaves no lock behind") unlock() } + +// A blocked record's redispatch re-runs what blocked it, and that rerun +// decides with a policy. The policy has to be the one the decision was +// authorized against: `Ledger.Redispatch` sets Rerun for a blocked record +// without consulting the served set at all, so a rerun carrying the file the +// command loaded at start-up can re-admit a record whose project the +// operator unserved in between — and write an admitted verdict that is not +// true of the policy as it stands (Copilot on #771). +func TestARerunDecidesWithThePolicyItWasAuthorizedAgainst(t *testing.T) { + const withdrawn, kept = int64(48699913), int64(48929974) + path, startup := policyFile(t, map[int64]admission.Project{kept: {}, withdrawn: {}}) + + unserved := startup + unserved.Projects = map[int64]admission.Project{kept: {}} + writePolicyFile(t, path, unserved) + + // What the command loaded at start-up still serves the project. + stale, err := startup.Policy(startup.Agent.PersonID) + require.NoError(t, err) + require.Contains(t, stale.Projects, withdrawn, "the stale policy is what would re-admit it") + + authorized, served, release, err := authorizedProfile(context.Background(), connectProfile{name: "agent", path: path, file: startup}) + require.NoError(t, err) + t.Cleanup(release) + + assert.Equal(t, []int64{kept}, served) + fresh, err := authorized.file.Policy(authorized.file.Agent.PersonID) + require.NoError(t, err) + assert.NotContains(t, fresh.Projects, withdrawn, + "the rerun decides against the reading the decision was authorized with, not the one the command started with") + assert.Contains(t, fresh.Projects, kept) +} + +// The redispatch's lock is held across the ledger's write and let go before +// the rerun, which talks to Basecamp. Both halves matter and neither is +// visible from the helper alone, so this asserts the order as a property of +// the source, the way the dispatcher's release point is asserted in +// connector. Released before the write and the window is open again; +// released after the rerun and an operator's command holds a lock `connect +// setup` waits on, across the network. +func TestTheRedispatchReleasesAfterTheWriteAndBeforeTheRerun(t *testing.T) { + source, err := os.ReadFile("connect_operator.go") + require.NoError(t, err) + body := string(source) + start := strings.Index(body, "func runConnectRedispatch(") + require.Positive(t, start, "runConnectRedispatch is where the order lives") + body = body[start:] + if end := strings.Index(body, "\nfunc "); end > 0 { + body = body[:end] + } + + authorize := strings.Index(body, "authorizedProfile(") + write := strings.Index(body, "ledger.Redispatch(") + release := strings.Index(body, "release()") + rerun := strings.Index(body, "rerunPrerequisite(") + require.Positive(t, authorize) + require.Positive(t, write) + require.Positive(t, release) + require.Positive(t, rerun) + + assert.Less(t, authorize, write, "the served set is read before the decision is written") + assert.Less(t, write, release, "and the lock is held across that write, not merely taken before it") + assert.Less(t, release, rerun, "and let go before the rerun, which reads Basecamp") +} diff --git a/internal/commands/connect_setup_test.go b/internal/commands/connect_setup_test.go index 961ac58d2..ac92509e2 100644 --- a/internal/commands/connect_setup_test.go +++ b/internal/commands/connect_setup_test.go @@ -830,11 +830,23 @@ func TestConnectSetupOperatorProfileFollowsEnvironmentPrecedence(t *testing.T) { } } +// shortSetupLockWait keeps a test that means to see contention from sitting +// out the production wait to see it: Lock waits for a holder now, and a test +// that inherits that bound measures the clock rather than the refusal +// (Copilot on #771). The bound itself is asserted where it lives, in setup. +func shortSetupLockWait(t *testing.T) { + t.Helper() + was := setup.LockWait + setup.LockWait = 20 * time.Millisecond + t.Cleanup(func() { setup.LockWait = was }) +} + // A profile another setup is working on is busy, not broken. func TestConnectSetupReportsAnotherSetupAsBusy(t *testing.T) { + shortSetupLockWait(t) s := startConnectSetupServer(t) connectSetupApp(t, s, "agent") - unlock, err := setup.Lock(connectSetupPath(t, "agent")) + unlock, err := setup.Lock(t.Context(), connectSetupPath(t, "agent")) require.NoError(t, err) t.Cleanup(unlock) @@ -1174,9 +1186,10 @@ func TestConnectSetupRefusesToRebindToAnotherIdentity(t *testing.T) { // Contention is retryable, and named as busy wherever it comes from. func TestConnectSetupBusyIsRetryable(t *testing.T) { + shortSetupLockWait(t) s := startConnectSetupServer(t) connectSetupApp(t, s, "agent") - unlock, err := setup.Lock(connectSetupPath(t, "agent")) + unlock, err := setup.Lock(t.Context(), connectSetupPath(t, "agent")) require.NoError(t, err) t.Cleanup(unlock) diff --git a/internal/connector/dispatch_policy_lock_test.go b/internal/connector/dispatch_policy_lock_test.go index d6d090789..3cd049185 100644 --- a/internal/connector/dispatch_policy_lock_test.go +++ b/internal/connector/dispatch_policy_lock_test.go @@ -1,6 +1,7 @@ package connector import ( + "context" "errors" "testing" "time" @@ -145,3 +146,34 @@ func TestALaunchSurvivesAnAuthorizeThatReturnsNoRelease(t *testing.T) { t.Fatal("no worker started") } } + +// The lock is held ACROSS the ledger's commit, not merely taken before it. +// Released a line early and every other test here still passes while the +// window is wide open again, so the release itself looks at the ledger: by +// the time it runs, the record this launch was authorized for has already +// moved to dispatched (Copilot's shape, and Codex on the same PR). +func TestThePolicyLockIsHeldAcrossTheLedgerCommit(t *testing.T) { + fake := newFakeDriver() + var h *dispatchHarness + committed := make(chan RecordState, 4) + h = newDispatchHarness(t, fake, func(o *DispatcherOptions) { + o.Authorize = func() (map[int64]admission.Project, func(), error) { + return map[int64]admission.Project{adapterBucketID: {}}, func() { + record, ok, err := h.ledger.Get(context.Background(), 1) + require.NoError(t, err) + require.True(t, ok) + committed <- record.State + }, nil + } + }) + admitOn(t, h.ledger, 1, "recording:1") + h.run(t) + + select { + case state := <-committed: + assert.Equal(t, StateDispatched, state, + "the release runs after the launch transaction, so an unserve waiting on the lock cannot land between the reading and the task") + case <-time.After(5 * time.Second): + t.Fatal("nothing was launched") + } +} diff --git a/internal/connector/dispatcher.go b/internal/connector/dispatcher.go index 073ef9c8e..00006506a 100644 --- a/internal/connector/dispatcher.go +++ b/internal/connector/dispatcher.go @@ -32,12 +32,24 @@ import ( // driver is asked for anything, a follow-up is exposed before its prompt // is sent, and an attempt is ended in the ledger only after its worker is // gone. -// 2. The project is connect.json's. A worker runs only for a record whose -// project connect.json still serves, and stops being handed follow-ups -// the moment it stops serving it. Where the worker runs is not the -// record's: every task runs in the directory the connector was started -// in, and a task that needs a clone or a directory of its own is the -// agent's business to make. +// 2. The project is connect.json's. A worker is started only for a record +// whose project connect.json serves when the launch commits: that set is +// read under the file's own lock and the lock is held across the commit, +// so an unserve lands wholly before the reading or wholly after the task +// exists. +// +// Follow-ups are held to the same set as it is read rather than as it is +// locked, so a task whose project stops being served stops being handed +// instructions within one reading of the file — the reader's TTL — and +// not instantly. That is deliberate: a follow-up runs on the task's own +// goroutine, where "the lock was busy" and "there is no more work" are +// the same answer, so contention would end tasks early. The task was +// authorized at launch; what the TTL bounds is how long one already +// running keeps being fed. +// +// Where the worker runs is not the record's: every task runs in the +// directory the connector was started in, and a task that needs a clone +// or a directory of its own is the agent's business to make. // 3. Nothing crosses to a worker that it does not need. The prompt names // events and a recording URL, never content, and is under // MaxPromptTokens at its worst case; the task token reaches only the @@ -1305,6 +1317,14 @@ func (r *taskRun) nextFollowUp(ctx context.Context) (int64, bool, error) { // Once, and the same set all the way down: the check, and the join it // authorizes. Read twice, the cache could turn over in between and the // two could disagree. + // + // The reader, not the lock the launch takes. Returning false here ends + // the task, so a lock another command holds for a moment would look + // exactly like a conversation with nothing left in it — a launch can + // give up its turn and try next tick, and this cannot. The task is + // already authorized; what is bounded here is how long one already + // running keeps being fed after an unserve, which is the reader's TTL + // (invariant 2). served, err := r.d.servedBuckets() switch { case err != nil: diff --git a/internal/connector/setup/private.go b/internal/connector/setup/private.go index f07a5f3d7..6b545f9d7 100644 --- a/internal/connector/setup/private.go +++ b/internal/connector/setup/private.go @@ -1,6 +1,7 @@ package setup import ( + "context" "errors" "fmt" "io" @@ -62,6 +63,26 @@ var ErrSetupRunning = errors.New("another command holds this profile's connector // guarantee is the lock, so it refuses rather than run without one. var ErrLockUnavailable = errors.New("this host cannot lock the connector's policy") +// Who takes this lock, and for how long. Two holders, and a third form for +// the one caller that must never wait: +// +// - `connect setup` (Lock) holds it across its whole load, change and +// save, network checks included. Seconds, and on a slow or unreachable +// account longer than that. +// - `connect redispatch` (Lock) holds it across one file read and one +// ledger transaction, then lets go before it re-runs a prerequisite +// against Basecamp. Milliseconds. It is a one-shot operator command, +// not the connector. +// - the running connector (TryLock) takes it once per launch, across one +// file read and one ledger transaction, and never waits: a pass that +// cannot take it dispatches nothing and asks again on its next tick. +// +// Nothing holds it across a network call except setup, and nothing takes any +// other lock while holding it — the ledger's SQLite locks are taken under +// it, never the other way about. A fourth caller that wants to wait on this +// while holding something the connector's launch path waits on is how this +// becomes a deadlock; there is no such caller today. +// // LockWait bounds Lock's wait for a holder to finish. The connector takes // this lock too now, and what it does under it is one file read and one // SQLite transaction — bounded by the ledger's own five-second busy retry — @@ -79,12 +100,27 @@ const lockPoll = 20 * time.Millisecond // Lock takes the per-profile setup lock beside connect.json, so a load, a // change and a save cannot interleave with another holder's. It waits up to -// LockWait and then reports ErrSetupRunning: the wait is for the connector's -// brief hold, not a queue of setups, and a second setup on one profile is -// still a mistake to report rather than to serialize — it will not have -// finished inside LockWait. -func Lock(path string) (unlock func(), err error) { - return lock(path, LockWait) +// LockWait, or until ctx ends, and then reports ErrSetupRunning or ctx's own +// error. +// +// The wait honors ctx, and a wait that ends takes nothing: a person who +// pressed Ctrl-C during contention is not made to sit out the bound, and a +// command whose context has expired does not go on to acquire a lock and act +// under it (Copilot on #771). What a caller that cannot get the lock gets is +// a refusal, never a fall-through: "the policy could not be checked" is not +// permission to change it. +// +// The wait is for the connector's hold, which is milliseconds. It cannot +// tell one holder from another, though, so a second `connect setup` that +// finishes inside LockWait is now queued rather than refused. That is a +// change: this used to refuse on sight, on the reasoning that a second setup +// on one profile is a mistake to report and not a queue. It is the price of +// not failing the first setup that happens to overlap a dispatcher pass, and +// the reasoning survives where it matters — a setup that is actually running +// takes far longer than LockWait, so a person who started two still gets +// told. +func Lock(ctx context.Context, path string) (unlock func(), err error) { + return lock(ctx, path, LockWait) } // TryLock takes the lock if it is free and reports ErrSetupRunning if it is @@ -93,11 +129,12 @@ func Lock(path string) (unlock func(), err error) { // one tick of latency. Waiting instead would park the dispatcher behind a // `connect setup`'s network checks, and a dispatcher that blocks on a file // another process writes is its own hazard. +// It takes no context because it does not wait: there is nothing to cancel. func TryLock(path string) (unlock func(), err error) { - return lock(path, 0) + return lock(context.Background(), path, 0) } -func lock(path string, wait time.Duration) (unlock func(), err error) { +func lock(ctx context.Context, path string, wait time.Duration) (unlock func(), err error) { if err := ensurePrivateDirs(path); err != nil { return nil, err } @@ -116,6 +153,12 @@ func lock(path string, wait time.Duration) (unlock func(), err error) { lock := flock.New(lockPath, flock.SetPermissions(0o600)) deadline := time.Now().Add(wait) for { + // Before the attempt, not only after it: a context that has already + // ended must not come away holding the lock and go on to act under + // it. + if err := ctx.Err(); err != nil { + return nil, err + } held, err := lock.TryLock() if err != nil { return nil, fmt.Errorf("%w: %s: %w", ErrLockUnavailable, lockPath, err) @@ -126,7 +169,11 @@ func lock(path string, wait time.Duration) (unlock func(), err error) { if !time.Now().Before(deadline) { return nil, fmt.Errorf("%w: %s", ErrSetupRunning, filepath.Dir(path)) } - time.Sleep(lockPoll) + select { + case <-ctx.Done(): + return nil, ctx.Err() + case <-time.After(lockPoll): + } } } diff --git a/internal/connector/setup/private_unix_test.go b/internal/connector/setup/private_unix_test.go index 328174949..b386c6119 100644 --- a/internal/connector/setup/private_unix_test.go +++ b/internal/connector/setup/private_unix_test.go @@ -3,6 +3,7 @@ package setup import ( + "context" "errors" "os" "path/filepath" @@ -187,14 +188,14 @@ func TestLockRefusesASecondSetup(t *testing.T) { shortLockWait(t) path, err := Path(configDir(t), "agent") require.NoError(t, err) - unlock, err := Lock(path) + unlock, err := Lock(context.Background(), path) require.NoError(t, err) - _, err = Lock(path) + _, err = Lock(context.Background(), path) assert.ErrorIs(t, err, ErrSetupRunning) unlock() - unlockAgain, err := Lock(path) + unlockAgain, err := Lock(context.Background(), path) require.NoError(t, err) unlockAgain() } @@ -235,7 +236,7 @@ func TestLockWaitsForAHolderThatFinishes(t *testing.T) { close(released) }() - got, err := Lock(path) + got, err := Lock(context.Background(), path) require.NoError(t, err, "the wait outlasts a holder that lets go") <-released got() @@ -265,3 +266,48 @@ func TestCheckPrivateFileCreatesNothingAndHoldsTheRules(t *testing.T) { require.NoError(t, os.Chmod(dir, 0o775)) require.ErrorIs(t, CheckPrivateFile(path), ErrNotPrivate, "a directory others can write") } + +// A wait that is canceled stops waiting and takes nothing. An operator who +// presses Ctrl-C during contention should not sit out LockWait, and a +// command whose context has already ended must not come away holding the +// lock and act under it: "the policy could not be checked" is a refusal, not +// permission (Copilot on #771). +func TestLockStopsWaitingWhenTheContextEnds(t *testing.T) { + path, err := Path(configDir(t), "agent") + require.NoError(t, err) + unlock, err := TryLock(path) + require.NoError(t, err) + t.Cleanup(unlock) + + // LockWait is left at its production value on purpose: a wait that + // ignored the context would take it, and this would take that long. + ctx, cancel := context.WithCancel(context.Background()) + go func() { + time.Sleep(5 * lockPoll) + cancel() + }() + + start := time.Now() + held, err := Lock(ctx, path) + assert.ErrorIs(t, err, context.Canceled) + assert.Nil(t, held, "a canceled wait comes away with nothing to release") + assert.Less(t, time.Since(start), LockWait/2, "and it stops when the context does, not when the bound expires") +} + +// A context that has already ended takes the lock from nobody, even when the +// lock is free: the caller is not going to act on what it reads. +func TestLockRefusesAnAlreadyEndedContext(t *testing.T) { + path, err := Path(configDir(t), "agent") + require.NoError(t, err) + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + held, err := Lock(ctx, path) + assert.ErrorIs(t, err, context.Canceled) + assert.Nil(t, held) + + // Proof that it refused rather than failed to acquire: the lock is free. + free, err := TryLock(path) + require.NoError(t, err) + free() +} diff --git a/skills/basecamp-connect/SKILL.md b/skills/basecamp-connect/SKILL.md index 986746393..0864099ad 100644 --- a/skills/basecamp-connect/SKILL.md +++ b/skills/basecamp-connect/SKILL.md @@ -121,7 +121,7 @@ the link and code while it waits; then wait for it to finish. |------|-------| | Credential | The CLI's credential store, under the profile. `basecamp auth status -P '' --json` describes it (see Inspecting). Never open it. | | connect.json | `$XDG_CONFIG_HOME/basecamp/connect//connect.json`, default `~/.config/basecamp/connect//connect.json`. Setup's JSON result gives the exact `path`. | -| Setup lock | `.connect.lock` beside connect.json. One setup per profile at a time. A running connector takes it too, for the moment it authorizes a launch or a redispatch against the file, so setup waits briefly rather than refusing when a dispatcher pass overlaps it. | +| Policy lock | `.connect.lock` beside connect.json. Three callers, and knowing which one holds it is how you read a `busy`. `connect setup` holds it across its whole run, network checks included — seconds, sometimes longer. `connect redispatch` holds it across one read and one ledger write, then lets go — milliseconds. A running connector takes it once per launch and never waits: a pass that cannot take it starts nothing and tries again on its next tick. Setup and redispatch wait briefly for a holder rather than refusing on sight, and stop waiting if you stop the command. | | Connector runtime state (ledger, checkpoint, lock) | `$XDG_STATE_HOME/basecamp/connect/-/`, default under `~/.local/state`; a shadow run's is under `connect-shadow/` instead. Read it only through `basecamp connect status` and `basecamp connect doctor`; never open or copy the files. | The CLI's configuration, its profiles and (when it uses files) its credential @@ -371,7 +371,7 @@ the exit status: exit 7 is shared. | `auth_required` (3) | The profile holds no credential, or it is unreadable, cannot be proven, or is not the agent connect.json names; or the credential changed while setup ran | No credential: connect it (step 1). Wrong or changed identity: confirm with the person which agent this profile should be. Changed mid-run: run setup again. | | `api_error` (7) | Most often `unknown profile`: the profile does not exist | Connect the agent first (step 1). | | `not_ready` (7) | A readiness check failed. `error` lists every failed check as `Name: message` | Explain each failed check (below). | -| `busy` (5) | Another command is using this profile's credential or setup lock — another setup, or a connector authorizing a launch | Nothing is wrong. Run setup again when it has finished. | +| `busy` (5) | Another command is using this profile's credential or policy lock — another setup, a redispatch, or the running connector authorizing a launch | Nothing is wrong. Run the command again when it has finished. | | `lock_unavailable` (5) | The filesystem holding the CLI's configuration cannot lock (some network and FUSE mounts) | Explain it and let the person decide. The fix is a local filesystem for `XDG_CONFIG_HOME`, and moving it hides every profile and stored file credential. After such a move do not reconnect the agent: that rotates its secret. | A setup the person stops also writes nothing. From e23423dbb9f96e925134502b7d447e8feb121e42 Mon Sep 17 00:00:00 2001 From: Jorge Manrubia Date: Sat, 19 Sep 2026 10:07:04 +0200 Subject: [PATCH 3/4] Fail closed when a launch holds no lock MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The guard added last round failed open. `Authorize` answering with a served set, no error and no release meant "the contract broke", and the dispatcher's response was to substitute a no-op and launch anyway — a task started against connect.json while nothing held it, recorded as authorized. That is the exact thing this branch exists to make impossible, reachable through the defensive branch written to make it safe, and the comment above it said the case could not happen while the code below quietly carried on when it did. It is ErrPolicyUnreadable now, and nothing launches. The wait's own cancellation had two races left. A context ending between the look and the flock came away holding the lock; a wait that ran out at the same moment its context ended reported the holder rather than the interruption. Both are checked now, and both are tested against a context that is live when the wait looks and over when it looks again — the race scheduled rather than hoped for. The redispatch's lock is one function with the release deferred inside it, so "held across the write, let go before the prerequisite" is structural rather than a matter of reading the command carefully. Its test is behavioral now, against a real connect.json and a real ledger on both paths out, and a separate one pins that the command binds the authorized profile back over its own, which every other test was blind to. Comments that stopped matching the code: the follow-up path is the reader's TTL and no longer claims otherwise in either place it is described; the contention error names all three callers rather than two; the lock-order note says what is actually taken under this lock, credential lock included, rather than claiming nothing is; and a rename left a dead symbol in a comment. What the rerun cannot promise is now written down rather than implied: it reads Basecamp, so it cannot run under the lock, and its verdict is taken against the policy the redispatch was authorized with. An unserve landing while the prerequisite runs can still be written as admitted — refused at the launch and named by the stranded report, the same residue as admission's own cached read. Co-Authored-By: Claude Opus 5 (1M context) --- internal/commands/connect_operator.go | 43 +++++++---- internal/commands/connect_policy_lock_test.go | 67 +++++++++++----- .../connector/dispatch_policy_lock_test.go | 35 ++++++--- internal/connector/dispatcher.go | 68 ++++++++++------ internal/connector/ledger_decisions.go | 8 +- internal/connector/setup/private.go | 33 ++++++-- internal/connector/setup/private_unix_test.go | 77 +++++++++++++++++++ 7 files changed, 252 insertions(+), 79 deletions(-) diff --git a/internal/commands/connect_operator.go b/internal/commands/connect_operator.go index c071cad7f..9ec1003eb 100644 --- a/internal/commands/connect_operator.go +++ b/internal/commands/connect_operator.go @@ -517,28 +517,14 @@ func runConnectRedispatch(cmd *cobra.Command, raw string) error { } defer done() - // The served projects as connect.json has them now, rather than the bit - // on the record: a record admitted while its project was served is not - // authorization to run it after the operator stopped serving it. Read - // under the setup lock and the lock held until the ledger has written, - // so an unserve lands wholly before the reading or wholly after the - // decision — not between them. // p itself is reassigned, so the reading the command started with is // gone rather than merely unused: everything below — the ledger's // decision and the rerun's policy alike — is the file as it was when // this was authorized. - p, served, release, err := authorizedProfile(ctx, p) + p, res, err := authorizedRedispatch(ctx, p, ledger, id) if err != nil { return err } - res, err := ledger.Redispatch(ctx, id, operatorName(), served) - // Let go before the prerequisite is re-run below: that talks to - // Basecamp, and nothing that waits on the network is held under a lock - // `connect setup` waits on. - release() - if err != nil { - return decisionError(err) - } report := connectRedispatchReport{RedispatchResult: res} if res.Worker != nil { // The ledger's own identity, not one assembled here: without @@ -559,6 +545,33 @@ func runConnectRedispatch(cmd *cobra.Command, raw string) error { return p.app.OK(report, output.WithSummary(redispatchSummary(report))) } +// authorizedRedispatch writes the decision under connect.json's own lock, +// against the served set read under it, and returns the profile carrying +// that reading. +// +// One function, and the release deferred inside it, because both halves of +// the claim should be structural rather than a matter of reading the command +// carefully: the lock is held across the ledger's write on every path, and +// it is let go before the caller re-runs a prerequisite — which talks to +// Basecamp, and must not happen under a lock `connect setup` waits on. A +// later edit to the command cannot leave it held, or take it early, without +// changing this (Codex on #771). +func authorizedRedispatch(ctx context.Context, p connectProfile, ledger *connector.Ledger, id int64) (connectProfile, connector.RedispatchResult, error) { + p, served, release, err := authorizedProfile(ctx, p) + if err != nil { + return p, connector.RedispatchResult{}, err + } + defer release() + // The served projects as connect.json has them now, rather than the bit + // on the record: a record admitted while its project was served is not + // authorization to run it after the operator stopped serving it. + res, err := ledger.Redispatch(ctx, id, operatorName(), served) + if err != nil { + return p, connector.RedispatchResult{}, decisionError(err) + } + return p, res, nil +} + func redispatchSummary(r connectRedispatchReport) string { var s string switch { diff --git a/internal/commands/connect_policy_lock_test.go b/internal/commands/connect_policy_lock_test.go index 245fb22de..599955ac4 100644 --- a/internal/commands/connect_policy_lock_test.go +++ b/internal/commands/connect_policy_lock_test.go @@ -6,6 +6,7 @@ import ( "log/slog" "os" "path/filepath" + "regexp" "strings" "testing" "time" @@ -271,34 +272,58 @@ func TestARerunDecidesWithThePolicyItWasAuthorizedAgainst(t *testing.T) { assert.Contains(t, fresh.Projects, kept) } -// The redispatch's lock is held across the ledger's write and let go before -// the rerun, which talks to Basecamp. Both halves matter and neither is -// visible from the helper alone, so this asserts the order as a property of -// the source, the way the dispatcher's release point is asserted in -// connector. Released before the write and the window is open again; -// released after the rerun and an operator's command holds a lock `connect -// setup` waits on, across the network. -func TestTheRedispatchReleasesAfterTheWriteAndBeforeTheRerun(t *testing.T) { +// The lock is held across the ledger's write and let go before the caller +// re-runs a prerequisite, on every path out. Both halves are what the shape +// buys: a release left to the command body reads fine and still leaks the +// lock across the network on the path somebody forgets. Asserted against a +// real connect.json and a real ledger rather than against the source, so it +// is the behavior and not the spelling (Codex on #771). +func TestTheRedispatchLetsGoOfTheLockOnEveryPath(t *testing.T) { + f := newOperatorFixture(t) + ledger := f.ledger(t, false) + t.Cleanup(func() { _ = ledger.Close() }) + path := connectSetupPath(t, "agent") + p := connectProfile{name: "agent", path: path, file: f.file} + + free := func(t *testing.T, why string) { + t.Helper() + unlock, err := setup.TryLock(path) + require.NoError(t, err, why) + unlock() + } + + // Record 2 is blocked, so this is the path that writes and then hands + // back a rerun for the caller to run against Basecamp. + authorized, res, err := authorizedRedispatch(context.Background(), p, ledger, 2) + require.NoError(t, err) + assert.True(t, res.Rerun, "a blocked record's prerequisite runs again") + assert.NotEmpty(t, authorized.file.Projects, "and the profile comes back carrying the reading it was authorized against") + free(t, "the lock is let go on the path that succeeds, before the prerequisite is re-run") + + // And on the path that refuses: an event id the ledger has never seen. + _, _, err = authorizedRedispatch(context.Background(), p, ledger, 9999) + require.Error(t, err) + free(t, "and on the path that refuses") +} + +// The command reassigns its own profile from the authorized one. Dropping +// that binding is invisible to every other test here and puts the rerun back +// on the file the command loaded at start-up, which is the defect this PR +// was reviewed for (Copilot on #771), so it is pinned as a property of the +// source. +func TestTheRedispatchCommandKeepsNoStartUpReading(t *testing.T) { source, err := os.ReadFile("connect_operator.go") require.NoError(t, err) body := string(source) start := strings.Index(body, "func runConnectRedispatch(") - require.Positive(t, start, "runConnectRedispatch is where the order lives") + require.Positive(t, start, "runConnectRedispatch is where the binding lives") body = body[start:] if end := strings.Index(body, "\nfunc "); end > 0 { body = body[:end] } - authorize := strings.Index(body, "authorizedProfile(") - write := strings.Index(body, "ledger.Redispatch(") - release := strings.Index(body, "release()") - rerun := strings.Index(body, "rerunPrerequisite(") - require.Positive(t, authorize) - require.Positive(t, write) - require.Positive(t, release) - require.Positive(t, rerun) - - assert.Less(t, authorize, write, "the served set is read before the decision is written") - assert.Less(t, write, release, "and the lock is held across that write, not merely taken before it") - assert.Less(t, release, rerun, "and let go before the rerun, which reads Basecamp") + call := regexp.MustCompile(`(\w+), \w+, \w+ :?= authorizedRedispatch\(`).FindStringSubmatch(body) + require.NotNil(t, call, "the command authorizes through authorizedRedispatch") + assert.Equal(t, "p", call[1], + "and binds the authorized profile back over p, so the start-up reading is out of scope rather than merely unused") } diff --git a/internal/connector/dispatch_policy_lock_test.go b/internal/connector/dispatch_policy_lock_test.go index 3cd049185..a3227e79d 100644 --- a/internal/connector/dispatch_policy_lock_test.go +++ b/internal/connector/dispatch_policy_lock_test.go @@ -127,10 +127,13 @@ func TestADispatcherNeedsThePolicyUnderItsLock(t *testing.T) { assert.Contains(t, err.Error(), "under its lock") } -// The launch defers Authorize's release, so a nil one would be a panic on the -// dispatcher's own path rather than a policy it could not read. Nothing in -// the tree returns one; the guard is for whatever wires this next. -func TestALaunchSurvivesAnAuthorizeThatReturnsNoRelease(t *testing.T) { +// An Authorize that answers with a served set, no error and no lock to hold +// authorizes nothing. This started life as a no-op release "so a wiring +// error would not panic the dispatcher", which is the same bug this PR is +// about wearing a different hat: no lock is held, the launch proceeds +// anyway, and the ledger records it as authorized. On this path a broken +// contract fails closed (Copilot on #771). +func TestAnAuthorizeWithNoLockToHoldLaunchesNothing(t *testing.T) { fake := newFakeDriver() h := newDispatchHarness(t, fake, func(o *DispatcherOptions) { o.Authorize = func() (map[int64]admission.Project, func(), error) { @@ -139,12 +142,26 @@ func TestALaunchSurvivesAnAuthorizeThatReturnsNoRelease(t *testing.T) { }) admitOn(t, h.ledger, 1, "recording:1") h.run(t) + time.Sleep(150 * time.Millisecond) - select { - case <-fake.made: - case <-time.After(5 * time.Second): - t.Fatal("no worker started") - } + fake.mu.Lock() + assert.Empty(t, fake.sessions, "a launch that holds no lock is not an authorized launch") + fake.mu.Unlock() + assert.Equal(t, StateAdmitted, getRecord(t, h.ledger, 1).State, "and the record is left for a pass that can take the lock") +} + +// And it is refused as the kind of nothing it is: nobody could take the +// lock, which is not the same as somebody holding it. +func TestAnAuthorizeWithNoLockToHoldIsUnreadableNotBusy(t *testing.T) { + fake := newFakeDriver() + h := newDispatchHarness(t, fake, func(o *DispatcherOptions) { + o.Authorize = func() (map[int64]admission.Project, func(), error) { + return map[int64]admission.Project{adapterBucketID: {}}, nil, nil + } + }) + _, _, err := h.d.authorizedBuckets() + assert.ErrorIs(t, err, ErrPolicyUnreadable) + assert.NotErrorIs(t, err, ErrPolicyBusy) } // The lock is held ACROSS the ledger's commit, not merely taken before it. diff --git a/internal/connector/dispatcher.go b/internal/connector/dispatcher.go index 00006506a..df7fa3df3 100644 --- a/internal/connector/dispatcher.go +++ b/internal/connector/dispatcher.go @@ -131,12 +131,20 @@ type DispatcherOptions struct { // launch, held across one SQLite transaction, and released before a // worker is spawned. // - // The release is never nil when the error is nil. A lock another - // process holds is reported as ErrPolicyBusy and authorizes nothing: - // the pass dispatches none and tries again on the next tick, rather - // than blocking behind a `connect setup`'s network checks. A lock or a - // file that could not be read at all is ErrPolicyUnreadable, which - // authorizes nothing either and says that something is wrong. + // An implementation that answers with a nil error and a nil release has + // broken its contract, and authorizes nothing: there is no lock to hold, + // so there is nothing to launch under. It is reported as + // ErrPolicyUnreadable rather than papered over — a no-op release + // substituted there would be an unlocked launch recorded as an + // authorized one, which is the exact thing this exists to prevent + // (Copilot on #771). + // + // A lock another process holds is reported as ErrPolicyBusy and + // authorizes nothing: the pass dispatches none and tries again on the + // next tick, rather than blocking behind a `connect setup`'s network + // checks. A lock or a file that could not be read at all is + // ErrPolicyUnreadable, which authorizes nothing either and says that + // something is wrong. Authorize func() (map[int64]admission.Project, func(), error) // TokenWindow is how long a task token's socket waits for the worker's // MCP server; DefaultTokenWindow when zero. @@ -475,10 +483,16 @@ func (d *Dispatcher) dispatchReady(ctx context.Context) error { served, servedErr := d.servedBuckets() // Follow-ups first: an event on a live conversation joins its task, while - // connect.json still serves that task's project. The served set goes to - // the ledger as well as being checked here, so what joins is held to the - // task's own project and to the set as it is now — not to the served bit - // admission wrote on each record when it decided it. + // connect.json still served that task's project as of this pass's + // reading. The served set goes to the ledger as well as being checked + // here, so what joins is held to the task's own project and to that + // reading — not to the served bit admission wrote on each record when it + // decided it. + // + // A reading, and one that may be up to connectServedTTL old: this is not + // the launch, and it is not under the lock. See invariant 2, and + // nextFollowUp, which shares this answer and where a busy lock and "no + // more work" would be the same answer. for _, r := range runs { // The same snapshot decides and is handed to the join. Reading it // again inside authorized would let the cache turn over between the @@ -621,21 +635,29 @@ func (d *Dispatcher) servedBuckets() ([]int64, error) { // between the read and the release, so the launch under it is decided against // the file as it is at the commit, not as it was when the pass began. // -// The release is never nil when the error is nil, and the caller lets go of -// it as soon as the ledger has committed: nothing that talks to a worker, -// a driver or the network happens under this lock. +// A release is what makes the reading worth anything, so this returns one or +// an error and never both-nothing: an Authorize that answers with neither is +// refused as ErrPolicyUnreadable rather than run without a lock. The caller +// lets go of it as soon as the ledger has committed — nothing that talks to +// a worker, a driver or the network happens under this lock. func (d *Dispatcher) authorizedBuckets() ([]int64, func(), error) { projects, release, err := d.opts.Authorize() - // Never nil either way, whatever Authorize returned: the caller defers - // the release, and a nil one would turn a policy that could not be read - // into a panic on the dispatcher's own path. - if release == nil { - release = func() {} - } if err != nil { - release() + // Something to call before returning, whatever Authorize left: the + // error is already the answer, and this only avoids a nil call on + // the way out. + if release != nil { + release() + } return nil, nil, err } + if release == nil { + // No error and nothing to hold. Something is wired wrong, and the + // safe reading of "no lock" is not "the lock is fine" — a no-op + // here would launch a task against connect.json while nothing held + // it and record that as authorized. Fail closed and say so. + return nil, nil, fmt.Errorf("%w: it answered with no lock to hold", ErrPolicyUnreadable) + } return d.narrow(projects), release, nil } @@ -1311,8 +1333,10 @@ func (r *taskRun) promptLoop(ctx context.Context, deadline, stillRunning <-chan } // nextFollowUp exposes the next event on the task not yet handed to the -// worker, and returns it. Nothing joins or is exposed once connect.json has -// stopped serving the task's project. +// worker, and returns it. Nothing joins or is exposed once the reader says +// connect.json has stopped serving the task's project: the reader, so within +// one of its readings of the file and not instantly — invariant 2, and the +// paragraph below. func (r *taskRun) nextFollowUp(ctx context.Context) (int64, bool, error) { // Once, and the same set all the way down: the check, and the join it // authorizes. Read twice, the cache could turn over in between and the diff --git a/internal/connector/ledger_decisions.go b/internal/connector/ledger_decisions.go index 121c7a8db..e01a816f0 100644 --- a/internal/connector/ledger_decisions.go +++ b/internal/connector/ledger_decisions.go @@ -191,10 +191,10 @@ func (l *Ledger) redispatch(ctx context.Context, eventID int64, by string, serve // to launch it is worse than one that refuses here (Copilot on #765). // // served is a reading its caller took under connect.json's own lock and - // holds until this commits (`connect redispatch`'s - // servedBucketsUnderLock), so an unserve lands wholly before that read - // or wholly after this write. This checks the set it was given; the - // caller is what makes the set current. + // holds until this commits (`connect redispatch`'s authorizedProfile, + // in internal/commands), so an unserve lands wholly before that read or + // wholly after this write. This checks the set it was given; the caller + // is what makes the set current. dispatchable := !record.ContentDropped && len(record.Decision.Snapshot) > 0 && record.Decision.Served && slices.Contains(served, record.BucketID) && record.Decision.ConversationKey != "" diff --git a/internal/connector/setup/private.go b/internal/connector/setup/private.go index 6b545f9d7..2872a6ecb 100644 --- a/internal/connector/setup/private.go +++ b/internal/connector/setup/private.go @@ -53,9 +53,11 @@ func ensurePrivateDirs(path string) error { } // ErrSetupRunning reports the per-profile policy lock held by somebody else. -// Two holders take it: `connect setup`, across its whole load-change-save, -// and a running connector, for the moment it authorizes a launch or a -// redispatch against the file. +// Three callers take it, and which one is holding decides whether a wait is +// worth anything: `connect setup` across its whole load-change-save, network +// checks included; `connect redispatch`, a one-shot operator command, across +// one read and one ledger write; and the running connector, for the moment +// it authorizes a launch. See the block above LockWait. var ErrSetupRunning = errors.New("another command holds this profile's connector policy") // ErrLockUnavailable reports a host that cannot take the setup lock at all: @@ -77,11 +79,12 @@ var ErrLockUnavailable = errors.New("this host cannot lock the connector's polic // file read and one ledger transaction, and never waits: a pass that // cannot take it dispatches nothing and asks again on its next tick. // -// Nothing holds it across a network call except setup, and nothing takes any -// other lock while holding it — the ledger's SQLite locks are taken under -// it, never the other way about. A fourth caller that wants to wait on this -// while holding something the connector's launch path waits on is how this -// becomes a deadlock; there is no such caller today. +// Setup is the only one that holds it across a network call. Other locks are +// taken UNDER this one and never the other way about: the ledger's SQLite +// locks on a redispatch and on a launch, and the credential key's lock on +// setup's final write. That one order is what keeps this from deadlocking, +// and a fourth caller that waits on this while holding something a current +// holder waits on is how it stops being one. There is no such caller today. // // LockWait bounds Lock's wait for a holder to finish. The connector takes // this lock too now, and what it does under it is one file read and one @@ -164,9 +167,23 @@ func lock(ctx context.Context, path string, wait time.Duration) (unlock func(), return nil, fmt.Errorf("%w: %s: %w", ErrLockUnavailable, lockPath, err) } if held { + // And again now it is held: a context that ended while this was + // acquiring must not come away holding the lock either, so it + // is given back rather than handed to a caller that is no + // longer entitled to act. + if err := ctx.Err(); err != nil { + _ = lock.Unlock() + return nil, err + } return func() { _ = lock.Unlock() }, nil } if !time.Now().Before(deadline) { + // A wait that ran out at the same moment the context ended is + // reported as the interruption it is: the caller stopped, and + // "somebody else holds it" would send them back to try again. + if err := ctx.Err(); err != nil { + return nil, err + } return nil, fmt.Errorf("%w: %s", ErrSetupRunning, filepath.Dir(path)) } select { diff --git a/internal/connector/setup/private_unix_test.go b/internal/connector/setup/private_unix_test.go index b386c6119..233b038da 100644 --- a/internal/connector/setup/private_unix_test.go +++ b/internal/connector/setup/private_unix_test.go @@ -311,3 +311,80 @@ func TestLockRefusesAnAlreadyEndedContext(t *testing.T) { require.NoError(t, err) free() } + +// endsWhileAcquiring is a context that is live when the wait looks before it +// tries, and over by the time it looks again with the lock in hand. It is +// the race a real cancellation runs — Ctrl-C landing in the microseconds +// between the check and the flock — scheduled rather than hoped for, since +// nothing else in a test can place a signal inside that window. +type endsWhileAcquiring struct { + context.Context + looks int +} + +func (c *endsWhileAcquiring) Err() error { + c.looks++ + if c.looks == 1 { + return nil + } + return context.Canceled +} + +// A context that ends while the lock is being acquired must not come away +// holding it: the caller is no longer entitled to act, and a lock handed to +// somebody who will not use it is a lock nobody else can take until the +// process exits (Codex on #771). +func TestLockGivesBackALockItAcquiredForAnEndedContext(t *testing.T) { + path, err := Path(configDir(t), "agent") + require.NoError(t, err) + + ctx := &endsWhileAcquiring{Context: context.Background()} + held, err := Lock(ctx, path) + assert.ErrorIs(t, err, context.Canceled) + assert.Nil(t, held) + require.Equal(t, 2, ctx.looks, "the wait looked before it tried and again once it held") + + free, err := TryLock(path) + require.NoError(t, err, "and gave back what it had taken: nothing is left holding it") + free() +} + +// The simpler half: a context that was already over before the call takes +// the lock from nobody at all. +func TestLockRefusesAnEndedContextWithoutTakingTheLock(t *testing.T) { + path, err := Path(configDir(t), "agent") + require.NoError(t, err) + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + held, err := Lock(ctx, path) + assert.ErrorIs(t, err, context.Canceled) + assert.Nil(t, held) + + free, err := TryLock(path) + require.NoError(t, err) + free() +} + +// A wait that runs out at the same moment its context ends reports the +// interruption rather than the holder: "somebody else has it" sends a person +// back to try again, and they stopped on purpose. +func TestAnExpiredWaitOnAnEndedContextReportsTheInterruption(t *testing.T) { + was := LockWait + LockWait = 0 + t.Cleanup(func() { LockWait = was }) + + path, err := Path(configDir(t), "agent") + require.NoError(t, err) + unlock, err := TryLock(path) + require.NoError(t, err) + t.Cleanup(unlock) + + // Live when the wait looks before trying, over by the time the bound + // has run out — so the two verdicts are available at the same moment + // and the interruption is the one reported. + ctx := &endsWhileAcquiring{Context: context.Background()} + _, err = Lock(ctx, path) + assert.ErrorIs(t, err, context.Canceled) + assert.NotErrorIs(t, err, ErrSetupRunning, "a person who stopped the command is not told to wait for somebody else") +} From 047938354b571de2259933a1229c3bce850ed95e Mon Sep 17 00:00:00 2001 From: Jorge Manrubia Date: Sat, 19 Sep 2026 10:17:53 +0200 Subject: [PATCH 4/4] Say what the lock does and does not promise MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit An adversarial pass over the finished branch found no hole left in the locking, and five places where the words around it were wrong. A comment that lies is the thing this change has been about since the first round, so they are worth their own commit. `connect redispatch`'s help now says what its re-run cannot promise, in the place a person reads before running it: the prerequisite talks to Basecamp, so it cannot hold the policy lock, and it decides against the served projects as they were when the redispatch was authorized. An unserve landing while it runs is not seen by that verdict. Nothing starts on it — the launch reads connect.json under the lock and refuses a project no longer served, and status counts what is left waiting. `rerunPrerequisite` said it decided "exactly as the connector's admission would", which is true of the rules and false of the reading; it now says which. `ErrPolicyBusy` said connect.json was being changed. A redispatch holding the same lock is writing SQLite and changing nothing, and the dispatcher cannot tell the two apart anyway — the lock is held, which is all it knows and all it needs to. `LaunchTask` named `connect redispatch` as one of the callers holding the lock across its transaction. Redispatch does that for its own write, which is `Ledger.Redispatch` and not this one. The note above `Lock` claimed a running setup takes far longer than the wait, so two setups would still be refused rather than queued. A setup can finish inside ten seconds. What is true is that queued setups each load, change and save under the lock in turn, which is what the lock is for. And one test claim: the helper test said the lock was held "until the decision has been written" while writing no decision. It now says what it checks, and a new source-level test carries the part only the source can answer — that the release is deferred before the ledger's write and taken nowhere else, which a test that only sees the lock free afterwards cannot tell from a release taken too early. A subtest name that had stopped being English is English again. Co-Authored-By: Claude Opus 5 (1M context) --- internal/commands/connect_operator.go | 27 +++++++++++++++---- internal/commands/connect_policy_lock_test.go | 27 ++++++++++++++++++- .../connector/dispatch_policy_lock_test.go | 2 +- internal/connector/dispatcher.go | 15 ++++++----- internal/connector/ledger_decisions.go | 2 +- internal/connector/ledger_tasks.go | 13 ++++----- internal/connector/setup/private.go | 16 +++++------ 7 files changed, 74 insertions(+), 28 deletions(-) diff --git a/internal/commands/connect_operator.go b/internal/commands/connect_operator.go index 9ec1003eb..66eaf7d12 100644 --- a/internal/commands/connect_operator.go +++ b/internal/commands/connect_operator.go @@ -465,8 +465,14 @@ A completed or held record is admitted at once (a completed one whose task is still running, when that task ends). A blocked record keeps its state and runs what blocked it again — the read, the events lookup, the served check — and is admitted the moment that succeeds; if it blocks again, the record stays -blocked with the authorization, and redispatch runs it again. While the hold -stands the record is authorized and nothing launches until release. +blocked with the authorization, and redispatch runs it again. That re-run +reads Basecamp, so it cannot hold the policy lock, and it decides against the +served projects as they were when the redispatch was authorized: an unserve +landing while it runs is not seen by that verdict. Nothing starts on it — the +connector reads connect.json under its lock at every launch and refuses a +project no longer served, and basecamp connect status counts what is left +waiting. While the hold stands the record is authorized and nothing launches +until release. It works on the ledger's transactions, so it is safe while the connector runs; the running connector dispatches what it admits.`, @@ -597,9 +603,20 @@ func redispatchSummary(r connectRedispatchReport) string { return s } -// rerunPrerequisite decides a blocked record again, as the agent, exactly as -// the connector's admission would: the verdict is revision-guarded, so a -// running connector deciding it at the same time is not a second verdict. +// rerunPrerequisite decides a blocked record again, as the agent, by the same +// rules the connector's admission applies: the verdict is revision-guarded, +// so a running connector deciding it at the same time is not a second +// verdict. +// +// By the same rules, and not against the same reading. The connector's +// admission asks a live reader for the served projects; this decides against +// the file as it was when the redispatch was authorized, because deciding +// reads Basecamp and nothing that waits on the network is done under a lock +// `connect setup` waits on. So an unserve completing while this runs is not +// seen here, and the verdict can admit a record whose project has just +// stopped being served. That is the deliberate residue, and it starts +// nothing: the launch reads connect.json under the lock and refuses it, and +// the stranded report names what is left waiting. func rerunPrerequisite(ctx context.Context, p connectProfile, ledger *connector.Ledger, id int64) (string, string, error) { agent, err := verifiedConnectAgent(ctx, p) if err != nil { diff --git a/internal/commands/connect_policy_lock_test.go b/internal/commands/connect_policy_lock_test.go index 599955ac4..93b654f17 100644 --- a/internal/commands/connect_policy_lock_test.go +++ b/internal/commands/connect_policy_lock_test.go @@ -199,7 +199,7 @@ func TestRedispatchReadsTheServedSetUnderTheLock(t *testing.T) { assert.NotContains(t, authorized.file.Projects, int64(48699913), "and gone from the profile everything below decides with") _, err = setup.TryLock(path) - assert.ErrorIs(t, err, setup.ErrSetupRunning, "and the lock is held until the decision has been written") + assert.ErrorIs(t, err, setup.ErrSetupRunning, "and it is still held when the reading comes back, for the caller to keep across its write") } // A setup that outlasts the wait leaves the redispatch unwritten, reported @@ -327,3 +327,28 @@ func TestTheRedispatchCommandKeepsNoStartUpReading(t *testing.T) { assert.Equal(t, "p", call[1], "and binds the authorized profile back over p, so the start-up reading is out of scope rather than merely unused") } + +// The release is deferred rather than called, so the lock is still held when +// the ledger writes. A test that only sees the lock free once the helper has +// returned cannot tell that apart from a release taken before the write, and +// that difference is the whole of the race (Codex on #771). This is the one +// claim about this path that only the source can answer. +func TestTheRedispatchHoldsTheLockAcrossTheLedgersWrite(t *testing.T) { + source, err := os.ReadFile("connect_operator.go") + require.NoError(t, err) + body := string(source) + start := strings.Index(body, "func authorizedRedispatch(") + require.Positive(t, start, "authorizedRedispatch is where the hold lives") + body = body[start:] + if end := strings.Index(body, "\nfunc "); end > 0 { + body = body[:end] + } + + deferred := strings.Index(body, "defer release()") + write := strings.Index(body, "ledger.Redispatch(") + require.Positive(t, deferred, "the release is deferred, so every path out takes it") + require.Positive(t, write) + assert.Less(t, deferred, write, "and deferred before the write, so the write happens under the lock") + assert.NotContains(t, body[deferred+len("defer release()"):], "release()", + "and nowhere else: a second release would let go before the write on some path") +} diff --git a/internal/connector/dispatch_policy_lock_test.go b/internal/connector/dispatch_policy_lock_test.go index a3227e79d..93b5ff6e2 100644 --- a/internal/connector/dispatch_policy_lock_test.go +++ b/internal/connector/dispatch_policy_lock_test.go @@ -71,7 +71,7 @@ func TestAPassAuthorizesNothingWhileThePolicyIsBusy(t *testing.T) { func TestAPassAuthorizesNothingWhenThePolicyCannotBeRead(t *testing.T) { for name, cause := range map[string]error{ "the policy could not be read under its lock": ErrPolicyUnreadable, - "something nothing here classified": errors.New("connect.json cannot be locked"), + "an otherwise unclassified error": errors.New("connect.json cannot be locked"), } { t.Run(name, func(t *testing.T) { fake := newFakeDriver() diff --git a/internal/connector/dispatcher.go b/internal/connector/dispatcher.go index df7fa3df3..5b6f23274 100644 --- a/internal/connector/dispatcher.go +++ b/internal/connector/dispatcher.go @@ -80,12 +80,14 @@ const MCPServerName = "basecamp" // ErrPolicyBusy is Authorize's answer when another process holds // connect.json's lock — a `connect setup` mid-run, which holds it across its -// network checks. It authorizes nothing and means nothing is wrong: the pass +// network checks, or a `connect redispatch` writing its decision. The file is +// not necessarily being changed; the lock is held, which is all this knows +// and all it needs to. It authorizes nothing and means nothing is wrong: the pass // dispatches none and asks again on the next tick, and so does every pass // until that setup lets go. A dispatcher that waited here instead would park // its hot path on a file another process writes, which is a worse thing to // own than a queue that drains the moment the policy is settled. -var ErrPolicyBusy = errors.New("connector: connect.json is being changed by another command") +var ErrPolicyBusy = errors.New("connector: connect.json's lock is held by another command") // ErrPolicyUnreadable is Authorize's answer when the served set could not be // taken under its lock at all: a host that cannot lock, a connect.json that @@ -539,15 +541,16 @@ func (d *Dispatcher) dispatchReady(ctx context.Context) error { // offer it again. continue case errors.Is(err, ErrPolicyBusy): - // Somebody is changing connect.json. Nothing is wrong and - // nothing is lost: this pass authorizes nothing and the next - // tick asks again. Every pass for as long as that setup runs + // Another command holds the policy lock — a setup changing + // connect.json, or a redispatch writing a decision against it. + // Nothing is wrong and nothing is lost: this pass authorizes + // nothing and the next tick asks again. Every pass for as long as that setup runs // gives up the same way, so the connector starts nothing while // its policy is being rewritten — which is the right thing to // do with a policy somebody is in the middle of changing, and // is cheaper than a dispatcher blocked on a file `connect // setup` holds across its network checks. - d.log.Debug("connector: connect.json is being changed; nothing is dispatched this pass") + d.log.Debug("connector: connect.json's lock is held by another command; nothing is dispatched this pass") return nil case errors.Is(err, ErrPolicyUnreadable): // Something is wrong, and the reader has already said so once. diff --git a/internal/connector/ledger_decisions.go b/internal/connector/ledger_decisions.go index e01a816f0..d151d13f8 100644 --- a/internal/connector/ledger_decisions.go +++ b/internal/connector/ledger_decisions.go @@ -191,7 +191,7 @@ func (l *Ledger) redispatch(ctx context.Context, eventID int64, by string, serve // to launch it is worse than one that refuses here (Copilot on #765). // // served is a reading its caller took under connect.json's own lock and - // holds until this commits (`connect redispatch`'s authorizedProfile, + // holds until this commits (`connect redispatch`'s authorizedRedispatch, // in internal/commands), so an unserve lands wholly before that read or // wholly after this write. This checks the set it was given; the caller // is what makes the set current. diff --git a/internal/connector/ledger_tasks.go b/internal/connector/ledger_tasks.go index 78b853684..6d3ffd371 100644 --- a/internal/connector/ledger_tasks.go +++ b/internal/connector/ledger_tasks.go @@ -331,12 +331,13 @@ func (l *Ledger) launchTask(ctx context.Context, spec LaunchSpec, attemptID stri // non-dispatcher caller fail closed. // // What closes the window between reading connect.json and - // committing this transaction is the caller's, and it is now held: - // Dispatcher.launch and `connect redispatch` read the served set - // under connect.json's own lock and keep the lock until this - // commits, so an unserve lands wholly before the reading or wholly - // after the task exists. This check is what that reading is spent - // on. + // committing this transaction is the caller's, and Dispatcher.launch + // — the only caller that reaches this one — now holds it: it reads + // the served set under connect.json's own lock and keeps the lock + // until this commits, so an unserve lands wholly before the reading + // or wholly after the task exists. This check is what that reading + // is spent on. (`connect redispatch` does the same thing for its own + // write, which is Ledger.Redispatch and not this.) return Launch{}, fmt.Errorf("connector: launch event %d in project %d, which is not served: %w", spec.EventID, record.BucketID, ErrNotStartable) } var busy bool diff --git a/internal/connector/setup/private.go b/internal/connector/setup/private.go index 2872a6ecb..7c81b9d87 100644 --- a/internal/connector/setup/private.go +++ b/internal/connector/setup/private.go @@ -114,14 +114,14 @@ const lockPoll = 20 * time.Millisecond // permission to change it. // // The wait is for the connector's hold, which is milliseconds. It cannot -// tell one holder from another, though, so a second `connect setup` that -// finishes inside LockWait is now queued rather than refused. That is a -// change: this used to refuse on sight, on the reasoning that a second setup -// on one profile is a mistake to report and not a queue. It is the price of -// not failing the first setup that happens to overlap a dispatcher pass, and -// the reasoning survives where it matters — a setup that is actually running -// takes far longer than LockWait, so a person who started two still gets -// told. +// tell one holder from another, though, so a second `connect setup` whose +// predecessor finishes inside LockWait is now queued rather than refused — +// two setups run one after the other where they used to be one setup and one +// refusal. That is a real change, and the price of not failing a setup that +// happens to overlap a dispatcher pass. Nothing is written twice by it: each +// one loads, changes and saves under the lock in turn, which is what the +// lock is for. A second setup whose predecessor is still going after +// LockWait is still refused. func Lock(ctx context.Context, path string) (unlock func(), err error) { return lock(ctx, path, LockWait) }