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 f906f3065..66eaf7d12 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,79 @@ func servedBucketsOf(file setup.File) []int64 { return served } +// 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 +// 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 authorizedProfile(ctx context.Context, p connectProfile) (connectProfile, []int64, func(), error) { + unlock, err := setup.Lock(ctx, p.path) + if err != nil { + return p, nil, nil, classifyDecisionLockError(p.name, err) + } + file, err := setup.Load(p.path) + if err != nil { + unlock() + 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 p, nil, nil, output.ErrUsage("connect.json now names another agent or account, so nothing was decided") + } + 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) { app := appctx.FromContext(cmd.Context()) if app == nil { @@ -75,7 +146,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. @@ -394,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.`, @@ -446,14 +523,13 @@ 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)) + // 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, res, err := authorizedRedispatch(ctx, p, ledger, id) if err != nil { - return decisionError(err) + return err } report := connectRedispatchReport{RedispatchResult: res} if res.Worker != nil { @@ -475,6 +551,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 { @@ -500,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 new file mode 100644 index 000000000..93b654f17 --- /dev/null +++ b/internal/commands/connect_policy_lock_test.go @@ -0,0 +1,354 @@ +package commands + +import ( + "context" + "encoding/json" + "log/slog" + "os" + "path/filepath" + "regexp" + "strings" + "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) + + // 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) +} + +// 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) + + 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 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 +// 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 = authorizedProfile(context.Background(), 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 := authorizedProfile(context.Background(), 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() +} + +// 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 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 binding lives") + body = body[start:] + if end := strings.Index(body, "\nfunc "); end > 0 { + body = body[:end] + } + + 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") +} + +// 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/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/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 new file mode 100644 index 000000000..93b5ff6e2 --- /dev/null +++ b/internal/connector/dispatch_policy_lock_test.go @@ -0,0 +1,196 @@ +package connector + +import ( + "context" + "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, + "an otherwise unclassified error": 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") +} + +// 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) { + return map[int64]admission.Project{adapterBucketID: {}}, nil, nil + } + }) + admitOn(t, h.ledger, 1, "recording:1") + h.run(t) + time.Sleep(150 * time.Millisecond) + + 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. +// 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 004b9dbd4..5b6f23274 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 @@ -66,6 +78,26 @@ 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, 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'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 +// 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 +119,35 @@ 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. + // + // 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. TokenWindow time.Duration @@ -207,6 +268,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 == "": @@ -422,10 +485,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 @@ -448,7 +517,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 +533,31 @@ 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): + // 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'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. + // 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 +615,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 +629,44 @@ 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. +// +// 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() + if err != nil { + // 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 +} + +// 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 +674,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 +763,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) @@ -1185,12 +1336,22 @@ 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 // 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/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..d151d13f8 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 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. 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..6d3ffd371 100644 --- a/internal/connector/ledger_tasks.go +++ b/internal/connector/ledger_tasks.go @@ -324,18 +324,20 @@ 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 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/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..7c81b9d87 100644 --- a/internal/connector/setup/private.go +++ b/internal/connector/setup/private.go @@ -1,11 +1,13 @@ package setup import ( + "context" "errors" "fmt" "io" "os" "path/filepath" + "time" "github.com/gofrs/flock" ) @@ -50,24 +52,98 @@ 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. +// 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: // 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. -func Lock(path string) (unlock func(), err error) { +// 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. +// +// 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 +// 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, 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` 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) +} + +// 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. +// It takes no context because it does not wait: there is nothing to cancel. +func TryLock(path string) (unlock func(), err error) { + return lock(context.Background(), path, 0) +} + +func lock(ctx context.Context, 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 +154,44 @@ 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 { + // 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) + } + 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 { + case <-ctx.Done(): + return nil, ctx.Err() + case <-time.After(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..233b038da 100644 --- a/internal/connector/setup/private_unix_test.go +++ b/internal/connector/setup/private_unix_test.go @@ -3,11 +3,13 @@ package setup import ( + "context" "errors" "os" "path/filepath" "syscall" "testing" + "time" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -173,21 +175,73 @@ 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) + unlock, err := Lock(context.Background(), path) require.NoError(t, err) - _, err = Lock(path) - assert.Error(t, err) + _, 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() } +// 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(context.Background(), 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)) @@ -212,3 +266,125 @@ 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() +} + +// 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") +} diff --git a/skills/basecamp-connect/SKILL.md b/skills/basecamp-connect/SKILL.md index 7308ffecb..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. | +| 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 | 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.