diff --git a/internal/commands/connect_operator.go b/internal/commands/connect_operator.go index ad4f71f19..11ed3e247 100644 --- a/internal/commands/connect_operator.go +++ b/internal/commands/connect_operator.go @@ -322,8 +322,11 @@ func renderConnectStatus(w io.Writer, r connectStatusReport) { for reason, n := range s.Blocked { fmt.Fprintf(w, " Blocked %s: %d\n", clean(reason), n) } + // A pending redispatch waits for its task to end. Only a build that ran + // workers wrote one or ended a task, so on this build the count is what + // an older one left, and none of them will run. if s.Review > 0 || s.AuthorizedBlocked > 0 || s.RedispatchPending > 0 { - fmt.Fprintf(w, " Review %d tagged, %d authorized and blocked, %d redispatches waiting for their task\n", s.Review, s.AuthorizedBlocked, s.RedispatchPending) + fmt.Fprintf(w, " Review %d tagged, %d authorized and blocked, %d redispatches left by an older build, which this one never runs\n", s.Review, s.AuthorizedBlocked, s.RedispatchPending) } fmt.Fprintf(w, "\n Live tasks %d\n", len(s.Tasks)) diff --git a/internal/commands/connect_run.go b/internal/commands/connect_run.go index 5c59e7034..73f32b6db 100644 --- a/internal/commands/connect_run.go +++ b/internal/commands/connect_run.go @@ -179,7 +179,7 @@ func runConnect(cmd *cobra.Command, f *connectRunFlags) error { if err != nil { return output.ErrUsage("The connector's state directory cannot be used: " + err.Error()) } - if err := connectLockFree(stateDir, account, file.Agent.PersonID); err != nil { + if err := connectLockFree(stateDir, account, file.Agent.PersonID, f.shadow); err != nil { return err } @@ -214,7 +214,7 @@ func runConnect(cmd *cobra.Command, f *connectRunFlags) error { return output.ErrAuth(err.Error()) } agentID := me.ID - lock, err := acquireConnectLock(stateDir, account, agentID) + lock, err := acquireConnectLock(stateDir, account, agentID, f.shadow) if err != nil { return err } @@ -402,18 +402,41 @@ func connectorStoppedBy(err error, agent, profile string, refused func() bool) ( // acquireConnectLock takes the connector's instance lock for the run, or // says another connector for this agent already holds it. -func acquireConnectLock(stateDir, account string, agentID int64) (*connector.InstanceLock, error) { +func acquireConnectLock(stateDir, account string, agentID int64, shadow bool) (*connector.InstanceLock, error) { lock, err := connector.AcquireInstanceLock(stateDir, account, agentID, time.Now()) - if errors.Is(err, connector.ErrAlreadyRunning) { - return nil, &output.Error{Code: output.CodeLockUnavailable, Message: err.Error()} + if running, ok := errors.AsType[*connector.AlreadyRunningError](err); ok { + return nil, errConnectorAlreadyRunning(running.Holder, shadow) } return lock, err } +// errConnectorAlreadyRunning is the refusal of a second connector for one +// agent. It names the holder, and says why a second one is never the answer: +// people reach for one to serve another repo, and the repo is not the +// connector's to choose. The session reading its lines picks the repo for +// each request's project, so one connector serves every repo the agent works +// in. +// +// A shadow run keeps its own state directory and lock (connectStateParts), so +// it never meets the connector's lock and the connector never meets its. A +// shadow refused here met another shadow, and is told so rather than told +// about connectors it does not compete with. +func errConnectorAlreadyRunning(holder string, shadow bool) *output.Error { + if shadow { + return &output.Error{Code: output.CodeLockUnavailable, + Message: "Another shadow run for this agent is already running: its lock is " + holder, + Hint: "A shadow run has its own state and lock, apart from the connector's, so only another shadow run holds this one. Stop it first."} + } + return &output.Error{Code: output.CodeLockUnavailable, + Message: "Another connector for this agent is already running: its lock is " + holder, + Hint: "One connector serves every repo: the session reading its requests picks the repo for each project, " + + "so an agent never needs a second connector. Use the session already running it, or stop that connector before starting another."} +} + // connectLockFree reports, by taking and at once releasing it, whether the // instance lock is free: the start's look before it waits for a token. -func connectLockFree(stateDir, account string, agentID int64) error { - lock, err := acquireConnectLock(stateDir, account, agentID) +func connectLockFree(stateDir, account string, agentID int64, shadow bool) error { + lock, err := acquireConnectLock(stateDir, account, agentID, shadow) if err != nil { return err } diff --git a/internal/commands/connect_run_test.go b/internal/commands/connect_run_test.go index 185875f9d..9b002b11c 100644 --- a/internal/commands/connect_run_test.go +++ b/internal/commands/connect_run_test.go @@ -2,6 +2,7 @@ package commands import ( "encoding/json" + "fmt" "log/slog" "os" "path/filepath" @@ -11,6 +12,7 @@ import ( "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + "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" @@ -157,3 +159,39 @@ func TestAShadowRunHasACheckpointLineageOfItsOwn(t *testing.T) { "and the connector's own lineage does not move") assert.NotEqual(t, connectConsumerNamespace(52007412, false), connectConsumerNamespace(52007412, true)) } + +// A second connector for an agent is refused with the holder named, and with +// why a second one is never needed: one connector serves every repo, because +// the session reading its requests picks the repo. A shadow run keeps its own +// lock, so a shadow refused met another shadow and is told only that. +func TestASecondConnectorIsToldOneServesEveryRepo(t *testing.T) { + for _, tc := range []struct { + name string + shadow bool + message string + hint string + notHint string + }{ + {name: "connector", message: "Another connector for this agent is already running", + hint: "One connector serves every repo", notHint: "shadow"}, + {name: "shadow", shadow: true, message: "Another shadow run for this agent is already running", + hint: "only another shadow run holds this one", notHint: "every repo"}, + } { + t.Run(tc.name, func(t *testing.T) { + dir := t.TempDir() + holder, err := connector.AcquireInstanceLock(dir, "2914079", 52007412, time.Now()) + require.NoError(t, err) + t.Cleanup(func() { _ = holder.Release() }) + + _, err = acquireConnectLock(dir, "2914079", 52007412, tc.shadow) + var refusal *output.Error + require.ErrorAs(t, err, &refusal) + assert.Equal(t, output.CodeLockUnavailable, refusal.Code) + assert.Equal(t, output.ExitRateLimit, output.ExitCodeFor(refusal.Code), "exit 5, as every lock refusal") + assert.Contains(t, refusal.Message, tc.message) + assert.Contains(t, refusal.Message, fmt.Sprintf("held by pid %d since ", os.Getpid()), "the refusal names the holder") + assert.Contains(t, refusal.Hint, tc.hint) + assert.NotContains(t, refusal.Hint, tc.notHint) + }) + } +} diff --git a/internal/commands/connect_start_wait_test.go b/internal/commands/connect_start_wait_test.go index 901e4c2df..f1b7d7ebd 100644 --- a/internal/commands/connect_start_wait_test.go +++ b/internal/commands/connect_start_wait_test.go @@ -518,7 +518,10 @@ func TestASecondConnectorIsRefusedBeforeItWaitsForAToken(t *testing.T) { select { case err := <-done: - assert.Equal(t, output.CodeLockUnavailable, usageError(t, err).Code, buf.String()) + refusal := usageError(t, err) + assert.Equal(t, output.CodeLockUnavailable, refusal.Code, buf.String()) + assert.Contains(t, refusal.Message, fmt.Sprintf("held by pid %d", os.Getpid()), "the refusal names the holder") + assert.Contains(t, refusal.Hint, "One connector serves every repo") case <-time.After(10 * time.Second): cancel() <-done diff --git a/internal/connector/admission/commit.go b/internal/connector/admission/commit.go index 95fab4a86..cf567a3ed 100644 --- a/internal/connector/admission/commit.go +++ b/internal/connector/admission/commit.go @@ -21,7 +21,7 @@ type Ledger interface { // - apply only when the record's revision is still v.Revision, the one // the decision was loaded at, and bump it; otherwise return // ErrAlreadyDecided. So an event is admitted at most once however many - // fetches, restarts or redispatches decide it, and a decision made on + // fetches, restarts or retries decide it, and a decision made on // an older load never overwrites a newer verdict, blocked ones // included; // - for an admitted verdict, read the conversation in the same @@ -133,18 +133,20 @@ func (k *keyedMutex) lock(ctx context.Context, key string) (func(), error) { // Blocked-record recovery. A record blocked on a read (read_failed, // read_unresolved), on unverified trust (trust_unverified) or on an // unverified assignment delta is retried every ten minutes for a day after it -// was first blocked, and on redispatch at any time after that. A throttled -// record (throttled) is on the same schedule, never before the server's -// deadline. A record blocked because connect.json could not be read -// (config_unreadable) is on the same interval with no window at all: that is -// a local file a person repairs, and an operator away for a week is -// ordinary, where a day of a failing server means the server is not coming -// back on its own. bucket_mismatch and unroutable are not timed: the -// pointer's bucket and type never change, so only a person's redispatch -// re-runs them. no_route is not timed either — it waits for the operator to -// serve the project, which is a decision, not a delay. A blocked record is -// never discarded for having failed: the checkpoint may already be past the -// event, and a tombstone would turn an outage into a permanent loss. +// was first blocked, and not after that. A throttled record (throttled) is on +// the same schedule, never before the server's deadline. A record blocked +// because connect.json could not be read (config_unreadable) is on the same +// interval with no window at all: that is a local file a person repairs, and +// an operator away for a week is ordinary, where a day of a failing server +// means the server is not coming back on its own. bucket_mismatch and +// unroutable are not timed: the pointer's bucket and type never change, so a +// re-run could only repeat the answer. no_route is not timed either: serving +// the project is a decision, not a delay. Nothing re-runs an untimed record, +// or one past its window. A person's redispatch did, and no command offers +// one since #815 removed it, so such a record stays blocked until a person +// discards it. A blocked record is never discarded for having failed: the +// checkpoint may already be past the event, and a tombstone would turn an +// outage into a permanent loss. // // The intake sweep is what runs this schedule (internal/connector, // Intake.sweepBlockedRetries): it asks the ledger for the records this @@ -173,11 +175,10 @@ var timedBlockedReasons = map[Reason]bool{ } // NextBlockedRetry returns when a blocked record should next be re-run, and -// false when it waits for something other than time: the operator serving the -// project (no_route), or a person's redispatch once the window has passed. -// notBefore is a throttled record's Verdict.RetryAt; no retry is scheduled -// before it, and a deadline past the window hands the record to redispatch -// rather than asking early. +// false when time will not settle it: an untimed reason (no_route among them), +// or a window that has passed. notBefore is a throttled record's +// Verdict.RetryAt; no retry is scheduled before it, and a deadline past the +// window ends the schedule rather than asking early. // // since is when the record entered the reason it is blocked on NOW, which is // not when it entered blocked. The reasons carry different windows, so a diff --git a/internal/connector/admission/doc.go b/internal/connector/admission/doc.go index 59ab0e35e..7ff0b5463 100644 --- a/internal/connector/admission/doc.go +++ b/internal/connector/admission/doc.go @@ -64,8 +64,8 @@ // retained and recovered. // 4. Only an admitted verdict carries the recording's content. A blocked or // discarded verdict carries the pointer, the reason, and for a blocked -// one what its holding reply needs. Nothing written to stdout carries -// content. +// one its trigger and reply destination. Nothing written to stdout +// carries content. // 5. One verdict per event, chosen and written in one ledger transaction. // The ledger refuses a verdict whose record changed after the decision // loaded it, and decides admitted or queued against the conversation as @@ -85,9 +85,10 @@ // // # What is not here // -// Tasks, attempts, the outbox, acknowledgements, and the holding reply a -// blocked(no_route) record receives are dispatch and lifecycle (plan steps 17 -// to 20). The boosted and queued triggers are step 25 and edited is step 28: -// each is a row added to the Matrix plus its rule, and the gate does not -// change. +// Handing a request off, acknowledging it, and replying. The connector writes +// each admitted request to stdout (internal/connector, RunHandoff), and the +// session reading it acknowledges and replies; a blocked(no_route) record gets +// no reply from either. The boosted and queued triggers are step 25 and edited +// is step 28: each is a row added to the Matrix plus its rule, and the gate +// does not change. package admission diff --git a/internal/connector/admission/matrix.go b/internal/connector/admission/matrix.go index c009978ae..f681edd12 100644 --- a/internal/connector/admission/matrix.go +++ b/internal/connector/admission/matrix.go @@ -20,10 +20,9 @@ type Rule struct { // on the assigner's say-so, so no broadened mode extends to them. OperatorOnly bool // RequiresServed discards the rule at the gate when connect.json does not - // serve the project. Only mentioned and assigned are answered in an - // unserved project (blocked(no_route) and a holding reply); every other - // trigger is discarded there, so it is cheaper to discard it before any - // read. + // serve the project. Only mentioned and assigned are kept in an unserved + // project (blocked no_route, and logged on stdout); every other trigger is + // discarded there, so it is cheaper to discard it before any read. RequiresServed bool // Acknowledge says a person asked for something, so the worker's // acknowledgement and the thirty-second guard apply. A completion or a @@ -133,30 +132,31 @@ const ( // named — moved since the event, most likely. ReasonBucketMismatch Reason = "bucket_mismatch" // ReasonUnroutable: the SDK has no typed read for the pointer's type. Not - // retried on a timer; only a redispatch re-runs it. + // retried on a timer, and nothing else re-runs it. ReasonUnroutable Reason = "unroutable" // ReasonNoRoute: shared by both states — a mentioned or assigned record in - // a project connect.json does not serve is blocked (and answered with a - // holding reply); any other trigger there is discarded. + // a project connect.json does not serve is blocked; any other trigger + // there is discarded. The blocked record is logged on stdout as an event + // line with this reason, and nothing replies to it: the connector posts + // nothing, and the session acts only on request lines. // // The stored value stays "no_route", the word a served project was called // by when it also named a directory. It is not renamed with the concept: - // it is written onto the record, and both the holding reply and the - // retraction that answers it read the record's reason back (askStillOpen, - // holdingReplyReason). A ledger in use carries rows and pending outbox - // intents written with this value, and a row is read as it was written. + // it is written onto the record, and a ledger in use carries rows and + // outbox intents, from when the connector posted holding replies, written + // with this value. A row is read as it was written. ReasonNoRoute Reason = "no_route" // ReasonConfigUnreadable: the record's answer turns on which projects // connect.json serves, and connect.json could not be read. Not // no_route: that says the operator has not served this project, which - // would be a false thing to say — and to post a holding reply about — - // when the truth is that nothing could read the file. + // would be a false thing to say when the truth is that nothing could read + // the file. // // It is the one blocked reason with no window: the intake sweep re-offers // it every BlockedRetryInterval for as long as it stands (NextBlockedRetry, // internal/connector Intake.sweepBlockedRetries). A day is the right bound // for a failing server, which is not coming back on its own after one; a // broken connect.json is a local file, and an operator away for a week is - // ordinary. `basecamp connect redispatch ` still runs one at once. + // ordinary. ReasonConfigUnreadable Reason = "config_unreadable" ) diff --git a/internal/connector/admission/run.go b/internal/connector/admission/run.go index 4d6c3ec1b..899253d75 100644 --- a/internal/connector/admission/run.go +++ b/internal/connector/admission/run.go @@ -22,8 +22,8 @@ type IDSource interface { } // Records loads a record for admission. It returns a record still being -// decided — seen, or blocked (recovery and redispatch decide a blocked record -// again) — with Event.Revision set to the record's revision as loaded. ok is +// decided (seen, or blocked, which the timed retry decides again) with +// Event.Revision set to the record's revision as loaded. ok is // false when the id is unknown or the record is past deciding, so it is // skipped rather than decided twice. // diff --git a/internal/connector/admission/verdict.go b/internal/connector/admission/verdict.go index a96b54c1d..02608a514 100644 --- a/internal/connector/admission/verdict.go +++ b/internal/connector/admission/verdict.go @@ -187,13 +187,10 @@ func WithSleep(sleep func(context.Context, time.Duration) error) Option { // WithServed makes the served projects live: admission reads them at each // decision instead of from the policy it was built with. // -// The dispatcher already rereads connect.json, so without this, serving a -// project while the connector runs changed only half the answer. Unserving -// one left its events admitted and then never started — no work and no -// holding reply, so the person who mentioned the agent got nothing. Serving -// one left them blocked no_route until a restart, and the holding reply's own -// remedy could not work: a redispatch re-runs admission, against the same -// stale policy (Copilot on #765). +// Without this, serving or unserving a project while the connector runs +// would not reach admission until a restart: a mention in a project just +// unserved would still be admitted, and one in a project just served would +// still be blocked no_route (Copilot on #765). // // Only the projects are live. Trust is not: who may drive the agent is a // different kind of decision, and changing it under a running connector is @@ -269,8 +266,7 @@ func (a *Admitter) Decide(ctx context.Context, ev Event) (out Verdict, err error // discarded: a discard is the one outcome repairing the file // cannot reverse, and a blocked record can still be run. It // comes round on the blocked schedule until the file is back, - // with no window (NextBlockedRetry), and a person may run - // `basecamp connect redispatch ` sooner. + // with no window (NextBlockedRetry). return v.end(StateBlocked, ReasonConfigUnreadable), nil } // Every other gate discard turns on the trust set, the matrix or the @@ -357,11 +353,11 @@ func (a *Admitter) Decide(ctx context.Context, ev Event) (out Verdict, err error v.address(summary) if !v.Served { - // Mentioned and assigned are answered in an unserved project rather - // than dropped: the record keeps its trigger and reply destination - // for the holding reply, and is read again once the operator serves - // the project. The gate already discarded every trigger that requires - // a served project. + // Mentioned and assigned are kept in an unserved project rather than + // dropped: the record is blocked with its trigger and reply + // destination, and its event line says why it was not handed off. + // Nothing runs it again. The gate already discarded every trigger + // that requires a served project. return v.end(StateBlocked, ReasonNoRoute), nil } v.State = StateAdmitted diff --git a/internal/connector/blocked_retry_test.go b/internal/connector/blocked_retry_test.go index 6217db7d7..dcba83973 100644 --- a/internal/connector/blocked_retry_test.go +++ b/internal/connector/blocked_retry_test.go @@ -476,12 +476,11 @@ func mustStamp(t *testing.T, s string) time.Time { } // Copilot on #770, in code the first round did not change: the queue carries -// an id, not the revision it was claimed at. `basecamp connect redispatch` -// runs beside a live connector (openConnectLedger takes the instance lock -// only for an import), so a person can re-decide the record after the sweep -// offered it and before admission took it. Admission would then load the -// newer revision and decide it at once, inside the interval the re-decision -// just wrote. +// an id, not the revision it was claimed at, so the record can be decided +// again after the sweep offered it and before admission took it (the +// redispatch command, removed in #815, did that beside a live connector). +// Admission would then load the newer revision and decide it at once, inside +// the interval the newer decision just wrote. // // Nothing decides a blocked record before its own schedule says so, whoever // offered it — so the stale hand-off is dropped where it is loaded. diff --git a/internal/connector/intake.go b/internal/connector/intake.go index d881c63fe..806a6ec54 100644 --- a/internal/connector/intake.go +++ b/internal/connector/intake.go @@ -897,8 +897,9 @@ const blockedRetryBatch = 100 // This is the whole of the blocked-record recovery schedule that // admission.NextBlockedRetry describes: without it that function computes a // time nothing acts on, and a read that failed during an outage, a throttle, -// or an unreadable connect.json waits for somebody to notice and run -// `basecamp connect redispatch ` on each one. +// or an unreadable connect.json is never decided again. Nothing else re-runs +// a blocked record: the hand-off connector has no command that does, so a +// record this schedule leaves alone stays blocked until a person discards it. // // The scope is the run's --project buckets, passed down rather than assumed. // A connector restricted to project A that offered a blocked record from diff --git a/internal/connector/ledger.go b/internal/connector/ledger.go index bf837272e..c06460ee3 100644 --- a/internal/connector/ledger.go +++ b/internal/connector/ledger.go @@ -42,8 +42,9 @@ const ( StateQueued RecordState = "queued" // StateBlocked is retained and retried: a reason, never a transport // failure dressed up as a verdict. Retried by the intake sweep, on - // admission.NextBlockedRetry's schedule (Intake.sweepBlockedRetries), and - // by a person at any time with `basecamp connect redispatch `. + // admission.NextBlockedRetry's schedule (Intake.sweepBlockedRetries). A + // record that schedule leaves alone stays blocked until a person discards + // it. StateBlocked RecordState = "blocked" // StateDispatched was handed to a worker. StateDispatched RecordState = "dispatched" @@ -551,8 +552,9 @@ END; // decided_at is when the latest verdict was written, blocked_at when the // record entered its current run of blocked verdicts (the retry window // counts from it), and retry_at a throttled verdict's server deadline. - // The rest is the verdict itself — what dispatch starts a task from, and - // what a blocked(no_route) record's holding reply needs. snapshot is the + // The rest is the verdict itself: what the handoff writes a request line + // from, and the trigger and reply destination a blocked record keeps (a + // worker-era build also started tasks from it). snapshot is the // recording's content as admission read it. Only an admitted verdict // writes one, whether the ledger writes it as admitted or as queued; a // record keeps it through dispatch and completion until retention drops diff --git a/internal/connector/ledger_admission.go b/internal/connector/ledger_admission.go index 62f41f0f4..ec828fab8 100644 --- a/internal/connector/ledger_admission.go +++ b/internal/connector/ledger_admission.go @@ -34,8 +34,7 @@ var ( func (l *Ledger) Admission() Admission { return Admission{ledger: l} } // undecided is where a record may be for admission to decide it: seen, which -// nothing has judged yet, and blocked, which recovery and redispatch decide -// again. Everything past these was decided by admission or moved on by +// nothing has judged yet, and blocked, which the timed retry decides again. Everything past these was decided by admission or moved on by // dispatch, and is never decided a second time. var undecided = []RecordState{StateSeen, StateBlocked} @@ -45,20 +44,20 @@ var undecided = []RecordState{StateSeen, StateBlocked} // // A blocked record is loaded only when something has given it an attempt to // make, and only once that attempt is due — whoever handed the id over. The -// queue carries an id and not the revision it was claimed at, and `basecamp -// connect redispatch` runs beside a live connector, so a person can re-decide -// a record between the sweep's offer and admission taking it; without this, -// admission would load the newer revision and decide it at once, inside the -// interval that re-decision had just written (Copilot on #770). +// queue carries an id and not the revision it was claimed at, so the record +// can be decided again between the sweep's offer and admission taking it; +// without this, admission would load the newer revision and decide it at +// once, inside the interval that newer decision had just written (Copilot on +// #770). // // No next_retry_at means no attempt is owed — an untimed reason, or a window // that has passed — and that is a reason to stay blocked, not permission to // run. Absent information is not consent: read the other way, a record with -// no schedule would be eligible immediately and on every tick after it. What -// gives an untimed record an attempt is a person: a redispatch stamps -// next_retry_at as it authorizes (authorizeBlocked), so the rerun it asks for -// runs at once and a rerun that never happened is picked up by the sweep -// rather than stranded. +// no schedule would be eligible immediately and on every tick after it. +// Nothing gives an untimed record an attempt today. A person's redispatch +// used to (Ledger.Redispatch stamps next_retry_at as it authorizes, through +// authorizeBlocked), but no command has called it since the worker side was +// removed in #815; such a record stays blocked until a person discards it. func (a Admission) LoadUndecided(ctx context.Context, id int64) (admission.Event, bool, error) { record, ok, err := a.ledger.Get(ctx, id) if err != nil || !ok { diff --git a/internal/connector/ledger_decisions.go b/internal/connector/ledger_decisions.go index e7f5afd6c..edfb8946a 100644 --- a/internal/connector/ledger_decisions.go +++ b/internal/connector/ledger_decisions.go @@ -12,8 +12,9 @@ import ( "github.com/basecamp/basecamp-cli/internal/connector/driver" ) -// ErrDecisionRefused is a redispatch or discard the record's state does not -// accept. The message says why. +// ErrDecisionRefused is a decision the record's state does not accept: a +// discard, or a redispatch through Ledger.Redispatch, which no command calls +// since #815. The message says why. var ErrDecisionRefused = errors.New("refused") // eventTask is the latest task an event was on, as a decision reads it. @@ -77,7 +78,8 @@ type operatorRecord struct { review bool authorizedAt sql.NullString // redispatchDecision is the redispatch waiting for the record's task to - // end; zero when none is. + // end; zero when none is. Only a ledger written before #815 removed the + // worker side, or Ledger.Redispatch, which no command calls now, holds one. redispatchDecision int64 } @@ -142,6 +144,9 @@ func (w LiveWorker) Identity() driver.Process { // Redispatch authorizes a record to run again, or for the first time, and // records who authorized it (invariants 4 to 6). // +// No command calls it: `connect redispatch` went with the worker side in +// #815. It stays with the schema it writes, which the ledger keeps as it is. +// // - completed with outcome unknown or failed: the task's token is // superseded; admitted at once when the task has ended, otherwise when it // ends. Refused without a snapshot, or in a project connect.json does @@ -190,11 +195,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 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. + // served must be a reading its caller took under connect.json's own lock + // and holds until this commits, so an unserve lands wholly before that + // read or wholly after this write; the removed `connect redispatch` did + // exactly that. 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 != "" @@ -468,8 +473,8 @@ WHERE event_id = ? AND state = 'pending' AND kind IN ('guard_ack', 'holding_repl // AuthorizedBlocked lists blocked records a person authorized, oldest first: // authorized since the record entered its current run of blocked states, so // an authorization that answered an earlier outcome does not count. -// The redispatch command runs the prerequisite itself; this would be for -// running it again when that did not settle it. Nothing calls it. The timed +// It was for running a redispatched record's prerequisite again when the +// redispatch's own run did not settle it. Nothing calls it. The timed // retry that now runs (Intake.sweepBlockedRetries) is not it: that schedule // is keyed on the reason, not on who authorized the record, and the reasons // it leaves alone — no_route, unroutable, bucket_mismatch — are the ones a @@ -514,13 +519,14 @@ func pendingNote(task eventTask) string { // (AuthorizedBlocked), and neither a move's own later stamp nor a clock that // stepped back may make a fresh one look stale. // It also makes the record due now, whatever it was blocked on. A person -// asking for a rerun is not a timer and does not wait for one: the redispatch -// command runs admission itself straight after this, and a blocked record -// with no attempt owed it is not one admission will load (LoadUndecided) — -// which is every untimed reason, no_route above all, the one a person is -// most likely to redispatch. Writing "due now" rather than leaving the -// schedule as it was is also what keeps a rerun that never happened — the -// command died between the two — on the sweep's list instead of stranding it. +// asking for a rerun is not a timer and does not wait for one: the removed +// redispatch command ran admission itself straight after this, and a blocked +// record with no attempt owed it is not one admission will load +// (LoadUndecided): every untimed reason, no_route above all, the one a +// person is most likely to redispatch. Writing "due now" rather than +// leaving the schedule as it was also keeps a rerun that never happened on +// the sweep's list instead of stranding it. Its only caller is +// Ledger.Redispatch, which no command calls since #815. func authorizeBlocked(ctx context.Context, tx *sql.Tx, eventID int64, now, by string) error { if _, err := tx.ExecContext(ctx, ` UPDATE events SET authorized_at = MAX(?, COALESCE(blocked_at, '')), authorized_by = ?, diff --git a/internal/connector/ledger_status.go b/internal/connector/ledger_status.go index 50caaf4ce..71b636fe9 100644 --- a/internal/connector/ledger_status.go +++ b/internal/connector/ledger_status.go @@ -103,6 +103,9 @@ type Status struct { Blocked map[string]int `json:"blocked"` // Review counts records tagged for review that have not reached a // person yet, and Authorized those a person authorized that have not run. + // RedispatchPending counts redispatches waiting for their task to end: + // only a build that ran workers (before #815) wrote one or ended a task, + // so on this build they are what an older one left, and never run. Review int `json:"review_tagged"` AuthorizedBlocked int `json:"authorized_blocked"` RedispatchPending int `json:"redispatch_pending"` diff --git a/internal/connector/ledger_tasks.go b/internal/connector/ledger_tasks.go index 7f6067ed4..83dfe164d 100644 --- a/internal/connector/ledger_tasks.go +++ b/internal/connector/ledger_tasks.go @@ -331,13 +331,12 @@ 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 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.) + // committing this transaction is the caller's: it has to 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. Dispatcher.launch was that caller, and nothing calls + // this since #815 removed the dispatcher. 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/lock.go b/internal/connector/lock.go index f153542b8..864c5ad63 100644 --- a/internal/connector/lock.go +++ b/internal/connector/lock.go @@ -15,6 +15,18 @@ import ( // ErrAlreadyRunning reports a second connector for the same agent. var ErrAlreadyRunning = errors.New("connector: another connector already holds this account and agent") +// AlreadyRunningError is ErrAlreadyRunning with who holds the lock, so a +// caller can word the refusal itself and still name the holder. +type AlreadyRunningError struct { + // Holder says who holds the lock: "held by pid 4242 since