Documentation
¶
Overview ¶
Package connector is the intake half of `basecamp connect`: the account event feed, the durable ledger behind it, and the recovery that makes a crash a delay rather than a loss.
What intake is, and what it deliberately is not ¶
Feed rows are pointers — id, type, bucket, creator, recording — and nothing else. No title, no body, no URL, no names. Intake writes that pointer to the ledger if the id is new and hands the id to a queue. That is the whole of the work on the feed's delivery path: a slow admission or a busy dispatcher is absorbed by the backlog rather than felt by the socket.
Absorbed, not unbounded. The backlog is a bounded queue, and at its pause threshold an offer waits for room — which is intake deliberately stopping its read of the feed, so the backlog cannot grow until it is the process's memory that fails. The decoupling holds up to that depth; past it, backpressure is the design, and the pause is reported rather than hidden.
Deciding whether an event deserves an agent's attention is admission's job and it costs a read per event it cares about. Intake does none of it. It also does no filtering of its own: the feed carries no `column_id` or `todolist_id`, so narrowing to a column or a list means re-fetching the parent listing, and that price belongs where the read already happens.
The two lanes ¶
The live lane is a WebSocket; the poll lane is an HTTP walk that runs about thirty seconds behind it, deliberately, so a page never stops mid-commit. The same event therefore arrives twice in the ordinary case, and dedupe is by event id across both lanes and across restarts — which is what makes the ledger, not a set in memory, the dedupe authority.
Only the poll lane advances the durable position. A live event id is far ahead of the poll lane, and a checkpoint taken from one would skip everything the safety delay had not yet served.
Quiet is not caught up ¶
Three separate things in this feed look like "nothing more to read" and are not:
- An empty page. A request crosses up to a thousand ledger rows and serves at most a hundred matches; rows the filters exclude still advance the cursor. Follow `next` until it is absent, never stop at the first empty page.
- A missing `next`. The walk reached its frozen head, which is not the end of history: a page cut short by the safety horizon withholds the link on purpose. Poll again later.
- A silent inbox. Delivery has write-time brakes — a per-account rate ceiling and an agent-to-agent chain breaker — and a braked event writes no addressing and tells the client nothing. A quiet feed is not proof of a quiet project, so nothing here ever reports "up to date".
Index ¶
- Constants
- Variables
- func AdoptableReply(c AdoptionCandidate, replies []AgentReply, lifecycle func(id int64) bool) (int64, bool)
- func ExitCodeForSignal(sig os.Signal) int
- func NotifyShutdown() (<-chan os.Signal, func())
- func ResolveStateDir(dir, accountID string) (string, int64, error)
- func RunAdmission(ctx context.Context, opts AdmissionOptions) error
- func RunHandoff(ctx context.Context, opts HandoffOptions) error
- func StateDirName(accountID string, agentID int64) string
- func StateRoot() (string, error)
- func StripMentionsOf(richText string, personID int64) string
- type Admission
- type AdmissionOptions
- type AdoptionCandidate
- type AgentReply
- type AttemptEnd
- type AttemptProcess
- type AttemptState
- type BlockedRetryScope
- type CommittedVerdict
- type Completion
- type ConnectionStatus
- type Decision
- type DecisionAction
- type Delivery
- type Destination
- type DiscardResult
- type DispatchStatus
- type DispatchedEvent
- type EntryClass
- type Gap
- type GapClass
- type GapStatus
- type HandoffLine
- type HandoffOptions
- type HandoffRecording
- type HeldStatus
- type Hold
- type HoldCause
- type HoldResult
- type HoldStatus
- type Hooks
- type ImportResult
- type InstanceHolderInfo
- type InstanceLock
- type Instruction
- type InstructionRecording
- type InstructionReply
- type Intake
- type Intent
- type IntentFilter
- type IntentKind
- type IntentResolution
- type IntentState
- type IntentStatus
- type Lane
- type Launch
- type LaunchSpec
- type Ledger
- func (l *Ledger) Admission() Admission
- func (l *Ledger) AdoptReply(ctx context.Context, taskID, eventID, replyID int64) error
- func (l *Ledger) AdoptionCandidates(ctx context.Context, taskID int64) ([]AdoptionCandidate, error)
- func (l *Ledger) AuthorizedBlocked(ctx context.Context, limit int) ([]int64, error)
- func (l *Ledger) Close() error
- func (l *Ledger) CloseLoss(ctx context.Context, lossID int64, now time.Time) (int, error)
- func (l *Ledger) CountInState(ctx context.Context, state RecordState) (int, error)
- func (l *Ledger) CreateTask(ctx context.Context, eventIDs []int64) (TaskGrant, error)
- func (l *Ledger) Discard(ctx context.Context, eventID int64, by string) (DiscardResult, error)
- func (l *Ledger) Dispatch(ctx context.Context, token string, agentID int64) (*TaskDispatch, error)
- func (l *Ledger) DropContent(ctx context.Context, discardedBefore, completedBefore time.Time) (int, error)
- func (l *Ledger) DueBlockedRetries(ctx context.Context, scope BlockedRetryScope) ([]Record, error)
- func (l *Ledger) EndAttempt(ctx context.Context, end AttemptEnd) (Settlement, error)
- func (l *Ledger) ExposeEvent(ctx context.Context, attemptID string, eventID int64) (bool, error)
- func (l *Ledger) ExtendLossDeadline(ctx context.Context, lossID int64, deadline time.Time) error
- func (l *Ledger) ForgetPosition(ctx context.Context, key eventfeed.CheckpointKey) error
- func (l *Ledger) Gaps(ctx context.Context) ([]Gap, error)
- func (l *Ledger) Get(ctx context.Context, id int64) (Record, bool, error)
- func (l *Ledger) Held(ctx context.Context) (bool, error)
- func (l *Ledger) HoldMarker(ctx context.Context) (Hold, bool, error)
- func (l *Ledger) Import(ctx context.Context, r Reconciliation, by string) (ImportResult, error)
- func (l *Ledger) Intent(ctx context.Context, id int64) (Intent, error)
- func (l *Ledger) Intents(ctx context.Context, f IntentFilter) ([]Intent, error)
- func (l *Ledger) IsLifecycleReceipt(ctx context.Context, kind MessageKind, id int64) (bool, error)
- func (l *Ledger) JoinConversation(ctx context.Context, taskID int64, served []int64) ([]int64, error)
- func (l *Ledger) LastPollServedID(ctx context.Context, key eventfeed.CheckpointKey) (int64, error)
- func (l *Ledger) LaunchTask(ctx context.Context, spec LaunchSpec) (Launch, error)
- func (l *Ledger) LineageHasPosition(ctx context.Context, key eventfeed.CheckpointKey) (bool, error)
- func (l *Ledger) LineagePollServedID(ctx context.Context, key eventfeed.CheckpointKey) (int64, error)
- func (l *Ledger) LiveAttempts(ctx context.Context) ([]LiveAttempt, error)
- func (l *Ledger) Load(ctx context.Context, key eventfeed.CheckpointKey) (string, bool, error)
- func (l *Ledger) MarkRunning(ctx context.Context, attemptID string, p AttemptProcess) error
- func (l *Ledger) MarkTakerUnaccounted(ctx context.Context, attemptID string) error
- func (l *Ledger) MarkUnrecoveredThrough(ctx context.Context, lossID, id int64) (int, error)
- func (l *Ledger) MissingIDs(ctx context.Context, lossID int64, state LossState) ([]int64, error)
- func (l *Ledger) NoteConnection(ctx context.Context, state, detail string) error
- func (l *Ledger) NotePollServed(ctx context.Context, key eventfeed.CheckpointKey, eventID int64) error
- func (l *Ledger) OpenLosses(ctx context.Context) ([]Loss, error)
- func (l *Ledger) RecordGap(ctx context.Context, gap Gap) (int64, error)
- func (l *Ledger) RecordLoss(ctx context.Context, droppedIDs []int64, now time.Time, window time.Duration, ...) (Loss, error)
- func (l *Ledger) RecordProgress(ctx context.Context, attemptID string) error
- func (l *Ledger) RecordRefusal(ctx context.Context, attemptID string) error
- func (l *Ledger) RecordSeen(ctx context.Context, ev eventfeed.Event, lane Lane) (bool, error)
- func (l *Ledger) RecordTaker(ctx context.Context, attemptID string, p AttemptProcess) error
- func (l *Ledger) RecordsInState(ctx context.Context, state RecordState, limit int) ([]Record, error)
- func (l *Ledger) RecordsInStateAfter(ctx context.Context, state RecordState, afterID int64, limit int) ([]Record, error)
- func (l *Ledger) Redispatch(ctx context.Context, eventID int64, by string, served []int64) (RedispatchResult, error)
- func (l *Ledger) Release(ctx context.Context, by string) (ReleaseResult, error)
- func (l *Ledger) ResolveIntent(ctx context.Context, id int64, r IntentResolution) error
- func (l *Ledger) Save(ctx context.Context, key eventfeed.CheckpointKey, position string) error
- func (l *Ledger) SaveRepairCursor(ctx context.Context, lossID int64, cursor string) error
- func (l *Ledger) SchemaVersion(ctx context.Context) (int, error)
- func (l *Ledger) SetHold(ctx context.Context, by string, cause HoldCause) (HoldResult, error)
- func (l *Ledger) SetHooks(h Hooks)
- func (l *Ledger) SetRepairEntry(ctx context.Context, lossID, since int64) error
- func (l *Ledger) SetState(ctx context.Context, id int64, state RecordState, reason string) error
- func (l *Ledger) StartableRecords(ctx context.Context, limit int) ([]Record, error)
- func (l *Ledger) StartableRecordsWhere(ctx context.Context, f StartableFilter) ([]Record, error)
- func (l *Ledger) Status(ctx context.Context) (Status, error)
- func (l *Ledger) StillRunning(ctx context.Context, attemptID string) (StillRunningTick, error)
- func (l *Ledger) StillScheduledBlockedIDs(ctx context.Context, scope BlockedRetryScope, ids []int64) ([]int64, error)
- func (l *Ledger) StrandedRecords(ctx context.Context, served []int64, buckets []int64) (int, error)
- func (l *Ledger) SupersedeTask(ctx context.Context, taskID int64) error
- func (l *Ledger) UnexposedEvents(ctx context.Context, taskID int64) ([]int64, error)
- func (l *Ledger) UnfinishedFromOthers(ctx context.Context, operatorID int64) (int, error)
- func (l *Ledger) UnrecoveredIDs(ctx context.Context) ([]int64, error)
- type LiveAttempt
- type LiveWorker
- type Loss
- type LossState
- type LossStatus
- type MembershipSource
- type MessageKind
- type Options
- type Outcome
- type Pointer
- type PositionStatus
- type PromoteOptions
- type PromoteResult
- type Queue
- type Receipt
- type Reconciliation
- type ReconciliationEntry
- type Record
- type RecordDecision
- type RecordState
- type RedispatchResult
- type ReleaseResult
- type Resolution
- type SDKMembership
- type SettledEvent
- type Settlement
- type StartableFilter
- type StateDirError
- type StateDirProblem
- type Status
- type StillRunningTick
- type StopReason
- type TaskDispatch
- type TaskGrant
- type TaskStatus
- type Tx
Constants ¶
const ( // ReasonHandedOff: the request was written to stdout for the session. ReasonHandedOff = "handed_off" // ReasonUnreadable: the request's record had no content to hand off, so // nothing was written. ReasonUnreadable = "unreadable" // ReasonBeforeThisRun: the request was made before this run started. // Nobody was there to take it, and it is not run late. ReasonBeforeThisRun = "before_this_run" )
Discard reasons the handoff writes.
const ( // DefaultRepairInterval is how often a repair walk re-runs while a loss is // open. It matches the feed's own repair poll, and it is deliberately // longer than the poll lane's ~30s safety delay: a walk that outran the // delay would call a still-committing event missing. DefaultRepairInterval = 60 * time.Second // DefaultRepairWindow is how long a loss stays open before the ids still // missing are called unrecovered. At the sixty-second cadence that is // about ten repair polls, each well past the safety delay. DefaultRepairWindow = 10 * time.Minute // DefaultMembershipInterval is how often the agent's project list is // re-read. The cable snapshots the agent's buckets when it subscribes, so // a project granted afterwards is invisible on the live lane until the // connection is remade. DefaultMembershipInterval = 10 * time.Minute )
Defaults for the recovery timers.
const ( // ReasonByOperator is a record a person closed without running it. ReasonByOperator = "by_operator" // ReasonImportedDone is a tombstone an import wrote for an entry a person // confirmed was finished before the cutover. ReasonImportedDone = "imported_done" )
Reasons a person's decision writes.
const ( // ConnectionRunning is a connector starting its parts — intake, // admission, dispatch, outbox — having passed every check before them. It // says nothing finer about the feed's socket, which intake does not report. ConnectionRunning = "running" // ConnectionStopped is a connector that exited through its own shutdown. // A second signal or a crash leaves the last state standing, so status // reads this beside the instance lock's holder rather than instead of it. ConnectionStopped = "stopped" // ConnectionDisconnected is a connector that stopped because Basecamp // refused the agent's credential: the agent was disconnected in Basecamp, // or connected on another computer. Its detail says so, for the person. ConnectionDisconnected = "disconnected" )
Connection states the run command reports for status.
const ( // DecisionDone is an entry a person confirmed the old connector's work // finished: the record becomes a tombstone. DecisionDone = "done" // DecisionHeld is every other mapped entry: the record is tagged for // review and waits for a person. DecisionHeld = "held" )
Reconciliation decisions.
const ( // DefaultBacklogWarn is the depth at which the backlog is worth saying out // loud. Nothing changes; the operator is told. DefaultBacklogWarn = 1_000 // DefaultBacklogPause is the depth at which intake stops consuming the // feed. The checkpoint does not move while it is paused, because the // package saves a position only after its page's events were accepted and // intake has stopped accepting them — so a crash while paused resumes from // before the backlog rather than after it. DefaultBacklogPause = 10_000 )
Backlog thresholds. Intake is the only work on the feed's delivery path, so the queue is where a slow admission shows up.
const ( // ExitInterrupted is 128 + SIGINT. ExitInterrupted = 130 // ExitTerminated is 128 + SIGTERM. ExitTerminated = 143 )
Exit codes for a signaled shutdown. 128+n is the shell's convention, and a supervisor reading the connector's exit status needs to tell an interrupt from a termination from a crash.
const AttemptIDLength = 4 + 24
AttemptIDLength is how long an attempt id is: "att_" and 12 random bytes in hex. Anything that has to know whether a path built from one fits (a unix socket's 103 bytes) asks here rather than guessing.
const DefaultHandoffInterval = 250 * time.Millisecond
DefaultHandoffInterval is how often the handoff looks for admitted records.
const HandoffGrace = time.Minute
HandoffGrace is how long before the run started a request may have been made and still be handed off: a mention written a moment before the session started is still meant for it.
const LedgerFile = "ledger.db"
LedgerFile is the ledger's file name inside the state directory.
const ReasonSpawnFailed = "spawn_failed"
ReasonSpawnFailed blocks an event whose worker could not be started a second time. It waits for a person's redispatch.
const ReconciliationVersion = 1
ReconciliationVersion is the reconciliation file format import reads.
const RefusedNote = "the request was refused; no message was created"
RefusedNote is the note on an intent Basecamp refused.
const StatusLimit = 20
StatusLimit is how many dispatches status lists.
Variables ¶
var ( // ErrTaskTokenRefused is a token that names no task, or one a // redispatch superseded. The two are not told apart: a worker holding // either has no task. ErrTaskTokenRefused = errors.New("the task token names no current task") // ErrNotOnTask is an event id the token's task does not carry. Whether // the event exists on another task is not said. ErrNotOnTask = errors.New("the event is not on this task") // ErrNotExposed is an acknowledgement or completion for an event the // worker was never handed. ErrNotExposed = errors.New("the event was not handed to this worker; call get_dispatch for it first") // ErrReportConflict is a second report that disagrees with the first. // Reported outcomes stand. ErrReportConflict = errors.New("the event already has a different report") // ErrNotDispatchable is an event whose record left the path to a worker // after it joined the task: blocked or withdrawn, or its content dropped. ErrNotDispatchable = errors.New("the event can no longer be dispatched") // ErrInvalidReport is a report the worker can correct: an outcome that // is not one of the two, a link that is not a URL, too many links. ErrInvalidReport = errors.New("the report is not valid") // ErrConversationBusy is an event whose conversation already has a // dispatched record outside the task being created: a running task, or // work a worker was handed that is not settled yet. A conversation has // one task at a time. ErrConversationBusy = errors.New("the event's conversation already has a task") // ErrNoSuchTask is a task id the ledger does not hold. ErrNoSuchTask = errors.New("no such task") // ErrEventOnLiveTask is an event a live task already carries. Handing it // to a second task would give two workers one instruction. ErrEventOnLiveTask = errors.New("the event is already on a live task") )
Refusals a worker's dispatch call can meet. Each is an answer, not a fault.
var ( // ErrNotStartable is a launch for a record that is not waiting for a // worker: not admitted or queued, without its snapshot, in a project // connect.json does not serve, or on a conversation that already has a // live task. ErrNotStartable = errors.New("the record is not waiting for a worker") // ErrNoLiveAttempt is a write for an attempt that has ended or never was. ErrNoLiveAttempt = errors.New("no live attempt by that id") )
Errors from the task ledger.
var ( // ErrNoSuchIntent is an intent id the ledger does not hold. ErrNoSuchIntent = errors.New("no such outbox intent") // ErrNotIndeterminate is a person's resolution for an intent that is not // indeterminate. ErrNotIndeterminate = errors.New("the intent is not indeterminate") // ErrReceiptOwned is a receipt another intent already owns. ErrReceiptOwned = errors.New("the receipt belongs to another intent") )
Errors from the outbox.
var ( // ErrNoShadowLedger is a shadow directory without a ledger. ErrNoShadowLedger = errors.New("there is no shadow ledger to promote") // ErrLedgerExists is a normal state directory that already has a ledger. ErrLedgerExists = errors.New("the connector already has a ledger") )
Errors from promote.
var ErrAlreadyRunning = errors.New("connector: another connector already holds this account and agent")
ErrAlreadyRunning reports a second connector for the same agent.
var ErrDecisionRefused = errors.New("refused")
ErrDecisionRefused is a redispatch or discard the record's state does not accept. The message says why.
var ErrHeld = errors.New("the connector is held")
ErrHeld is the hold marker refusing to hand a worker something new. It is not a failure of the task: nothing more is handed over until release.
var ErrHeldByWorker = errors.New("a worker was handed this event; it leaves dispatched only when completed")
ErrHeldByWorker is a move out of dispatched for an event a worker was handed and has not reported on. Only its outcome moves it.
var ErrLedgerNotTheSameFile = errors.New("the ledger path no longer names the file this process checked")
Opening one ledger file more than once in a process, safely.
The privacy check opens the file and closes it, and POSIX drops every lock a process holds on a file when any descriptor for it is closed — including the locks SQLite is holding on another connection. So the check runs exactly once per file per process, while nothing else has it open. A later Ledger on the same file (a status read beside a running connector, a promote) is verified instead against what that check established: the same file, still this user's own, still owner-only, in a directory that is still 0700. Stat never opens anything, so it takes no locks away.
ErrLedgerNotTheSameFile is a second open of a path that no longer names the file the check passed.
var ErrLedgerOutOfDate = errors.New("the ledger is older than this build; start the connector once to bring it up to date")
ErrLedgerOutOfDate is a ledger an older binary wrote, which this build has not migrated. Starting the connector migrates it.
var ErrLedgerSchema = errors.New("the connector ledger's schema is not the version this basecamp writes")
ErrLedgerSchema is a ledger whose schema is not the one this binary writes.
var ErrLedgerUnderAnotherName = errors.New("this process already has this ledger open under another name")
ErrLedgerUnderAnotherName is an open of a file this process already has open under a different path — a hardlink, or a route through a symlink. SQLite names its write-ahead log and shared-memory files after the path it was given, so one file opened under two names is two different logs for one database. It is refused here, before the check that would open the file and drop the live handle's locks.
var ErrNoSuchRecord = errors.New("no such event record")
ErrNoSuchRecord reports a state change addressed at an id the ledger does not hold.
var ErrNotAStateDir = errors.New("not the connector's state directory for this account")
ErrNotAStateDir is a directory that is not a connector state directory for the account asked about. StateDirError carries why.
var ErrNotATransition = errors.New("not a transition the ledger's lifecycle allows")
ErrNotATransition reports a state change the lifecycle does not have.
var ErrNotOnALiveTask = errors.New("no live task carries this event; a record is dispatched by creating a task for it")
ErrNotOnALiveTask is a move into dispatched for a record no live task carries. Records are dispatched by creating a task for them.
var ErrOnALiveTask = errors.New("a live task still carries this event; supersede the task first")
ErrOnALiveTask is a move out of dispatched, other than to completed, for a record a live task still carries. Its task is superseded first.
Functions ¶
func AdoptableReply ¶
func AdoptableReply(c AdoptionCandidate, replies []AgentReply, lifecycle func(id int64) bool) (int64, bool)
AdoptableReply applies the adopted-reply rule: exactly one reply by the agent at the destination after the event's acknowledgement, not after a later instruction's acknowledgement, and not one of the connector's own lifecycle messages.
func ExitCodeForSignal ¶
ExitCodeForSignal maps a shutdown signal to the exit code the connector leaves behind. An unknown signal reports 1: the process ended, and pretending it ended cleanly would be a lie to whatever restarts it.
func NotifyShutdown ¶
NotifyShutdown returns a channel carrying shutdown signals, and a stop function. Separated from the exit-code mapping so the mapping can be tested without sending real signals to the test binary.
The channel holds two: the first asks for an orderly shutdown, and the second is a person who has waited long enough. A caller that takes only the first leaves the second in the buffer, where it would be dropped rather than heard, which is why the buffer is two and the run reads both.
func ResolveStateDir ¶
ResolveStateDir is the one place a state directory is accepted: dir must be exactly StateRoot/<account>-<agent person id>, and its account must be accountID, compared as numbers. It returns the directory made absolute — which is the one every caller should go on to use — and the agent's Person id.
The location is part of the check, not only the name. A directory named for this account anywhere else — a copy of another account's ledger renamed to match — is refused, because the name is what binds a ledger to an account and anyone can choose a name.
func RunAdmission ¶
func RunAdmission(ctx context.Context, opts AdmissionOptions) error
RunAdmission takes the ids intake hands over and decides each against the ledger, until ctx ends. It returns what admission.Run returns.
func RunHandoff ¶
func RunHandoff(ctx context.Context, opts HandoffOptions) error
RunHandoff writes every admitted record as an event line until ctx ends.
func StateDirName ¶
StateDirName is the connector's state directory for one account and agent, "<account>-<agent person id>", inside StateRoot.
func StateRoot ¶
StateRoot is where every connector state directory lives: $XDG_STATE_HOME/basecamp/connect, or ~/.local/state/basecamp/connect when XDG_STATE_HOME is unset or not absolute, as the XDG specification says.
func StripMentionsOf ¶
StripMentionsOf removes every mention of personID from rich text, and leaves the rest as it was. A worker handed its own mention reads an instruction addressed to itself, which says nothing the dispatch does not already say.
What counts as a mention is exactly what basecamp.MentionedPersonIDs — the reader admission decided the trigger with — counts: the tag walk below is that reader's, rule for rule (comments, "<!", "<?" and end tags skipped, punctuation part of a name, the first sgid attribute authoritative even when empty, entities decoded, an unterminated tag ending the markup). The two are held to agreement by a differential test and a fuzz target over hostile markup, not by review.
A mention element runs from its start tag to the first end tag of the same name, unless it closes itself, another attachment starts first, or none closes, in which case the start tag stands alone. So no removal can swallow the instruction between a self-closing mention and some later stray closing tag. What it leaves behind is a space, not nothing: closing the gap could join a "<" before the element to the text after it into a tag that swallows what follows — someone else's mention included — and a space can never begin one.
Types ¶
type Admission ¶
type Admission struct {
// contains filtered or unexported fields
}
Admission is the ledger as admission sees it: the Records it loads events from and the Ledger its verdicts are committed to.
Admission is its own package and imports nothing from this one. This is the whole of the join: the pointer intake recorded goes out as an admission Event with the revision it was loaded at, and a verdict comes back through the ledger's one state-changing write, so the lifecycle that keeps a finished record finished is the same lifecycle a verdict meets.
func (Admission) Commit ¶
Commit writes a verdict onto its record in one transaction, and returns the state it wrote. It holds admission.Ledger's contract:
- the write applies only while the record is still at the verdict's revision and still undecided, and it bumps the revision; otherwise it is admission.ErrAlreadyDecided;
- an admitted verdict is written as queued when its conversation is live, decided inside the transaction: another record on the key is admitted and not yet dispatched (it becomes the task) or dispatched (the task is running). A queued record alone is not a live conversation: queued records join their task before it closes, and counting one left behind would queue every later event on the key with nothing to start a task;
- content is written only with an admitted verdict;
- a throttled verdict keeps its RetryAt.
func (Admission) LoadUndecided ¶
LoadUndecided loads a seen or blocked record as the event admission decides, with the revision it was loaded at. An unknown id, or a record past deciding, is not ok and is skipped.
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).
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.
type AdmissionOptions ¶
type AdmissionOptions struct {
Ledger *Ledger
Queue *Queue
Admitter *admission.Admitter
// Workers is the fetcher pool size; admission's default when zero.
Workers int
// Lines is the writer intake's pointer lines go through. Verdict lines
// share it, so the two never tear each other on one stdout.
Lines *ndjson.Writer
Logger *slog.Logger
}
AdmissionOptions wires admission onto intake's queue and ledger.
type AdoptionCandidate ¶
type AdoptionCandidate struct {
TaskID int64
EventID int64
ReplyKind string
ReplyRecordingID int64
// DeliveredAt is the event's ack_dispatch.
DeliveredAt time.Time
// NextAckAt is the first acknowledgement of a later instruction on the
// CONVERSATION, which may be on a task started after this one ended;
// zero when there is none.
NextAckAt time.Time
// AckID is the worker's own acknowledgement, which is never its reply
// however the clocks compare.
AckID int64
}
AdoptionCandidate is an event whose worker's report was lost after it acknowledged: settled unknown, delivered, and with no reply of its own.
type AgentReply ¶
AgentReply is a comment or chat line by the agent at a destination.
type AttemptEnd ¶
type AttemptEnd struct {
AttemptID string
Stop StopReason
// SpawnFailed is the driver's report that no worker process ever existed
// (driver.ErrNotStarted). Nothing else makes an exposure withdrawable.
SpawnFailed bool
// NoAutomaticRetry refuses the withdrawal even then: a task under the
// sandbox launcher is never retried automatically.
NoAutomaticRetry bool
// UnrecordedRefusals are refusals RecordRefusal could not write when they
// happened, settled here with the attempt. Refusals it did write are
// already on the attempt.
UnrecordedRefusals int
}
AttemptEnd is how an attempt ended.
type AttemptProcess ¶
type AttemptProcess struct {
PID int
PGID int
StartedAt time.Time
// StartedExact says StartedAt is the kernel's own start time for the pid
// (driver.Process.StartedExact). Only then is it written.
StartedExact bool
SessionID string
}
AttemptProcess is what MarkRunning records: the worker's process, where there is one, and its session id.
Only an identity the kernel gave is written. A start time the connector guessed cannot tell a pid from a later process that reused it, so it is left out of the record rather than written as if it could, and a record with a pid and no start time is one a later process signals nothing on (driver.ErrIdentityUnknown). What the ledger holds is therefore exact by construction, which is why Identity reads it back as exact.
func (AttemptProcess) Identity ¶
func (p AttemptProcess) Identity() driver.Process
Identity is the process the record names, for the one-owner rule. A start time in the ledger is the kernel's, since nothing else is written.
type AttemptState ¶
type AttemptState string
AttemptState is where an attempt is.
const ( // AttemptLaunching is written before the driver is asked to start a // worker. Found after a crash it is treated as running: the worker may // exist. AttemptLaunching AttemptState = "launching" // AttemptRunning has its process or session id. AttemptRunning AttemptState = "running" // AttemptEnded has a stop reason. AttemptEnded AttemptState = "ended" )
type BlockedRetryScope ¶
type BlockedRetryScope struct {
// Buckets is the run's --project scope — the same slice admission's
// policy gates on, with the same meaning: empty is a run restricted to no
// project, which is every project the agent can see, NOT no project at
// all.
//
// It is not optional, and it is not a detail. out_of_scope is a terminal
// discard, and the projects a run leaves out are another run's to
// dispatch: a sweep that offered a record from outside its scope would
// cause the permanent loss the retry exists to prevent. Every other
// terminal verdict a re-offered record can reach — stale, not_addressed,
// untrusted_performer, agent_authored — is the verdict a fresh event
// would get, and is correct.
Buckets []int64
// Now is the moment the sweep is asking about.
Now time.Time
// Limit is the most records one page of the query returns.
Limit int
// AfterRetryAt and AfterID are one cursor, in the order the query returns
// rows: the next page is what sorts after them. The caller pages because
// only it knows which rows it has already claimed, and a page that came
// back full of claims would otherwise spend a sweep's whole budget on
// work nothing can do.
//
// The cursor is the due time first and the id second because that is the
// order events_next_retry holds, and paging in any other order is what
// makes the query ignore it. Their zero values are before every row.
AfterRetryAt time.Time
AfterID int64
}
BlockedRetryScope narrows DueBlockedRetries to what this run may decide.
type CommittedVerdict ¶
type CommittedVerdict struct {
EventID int64
State RecordState
Reason string
Trigger string
Acknowledge bool
ReplyKind string
// ReplyRecordingID is where a reply to the event goes.
ReplyRecordingID int64
}
CommittedVerdict is what VerdictCommitted is told.
type Completion ¶
Completion is a worker's report for one event.
type ConnectionStatus ¶
type ConnectionStatus struct {
State string `json:"state"`
PID int `json:"pid"`
ChangedAt time.Time `json:"changed_at"`
Detail string `json:"detail,omitempty"`
}
ConnectionStatus is the run command's own record of its last run: running as its parts start, stopped when it exited. It is not the feed socket's state, which intake does not report.
type Decision ¶
type Decision struct {
// DecidedAt is when the latest verdict was written.
DecidedAt *time.Time
// BlockedAt is when the record entered its current run of blocked
// verdicts; nil unless it is blocked.
BlockedAt *time.Time
// RetryAt is a throttled verdict's server deadline; nil otherwise.
RetryAt *time.Time
// RetrySince is when the record entered the reason it is blocked on now
// — not when it entered blocked, which is BlockedAt. The retry window is
// the reason's, so it counts from here.
RetrySince *time.Time
// NextRetryAt is when the schedule next owes this record an attempt, as
// admission.NextBlockedRetry answered when the verdict was written. Nil
// when it owes none: an untimed reason, a window that has passed, or any
// state but blocked.
NextRetryAt *time.Time
Trigger string
Acknowledge bool
ConversationKey string
ReplyKind string
ReplyRecordingID int64
// Served is whether connect.json served the record's project when
// admission decided it.
Served bool
Class string
RecordingURL string
RequesterID int64
// Snapshot is the recording's content as admission read it, JSON. An
// admitted verdict writes it, as admitted or queued; it stays through
// dispatch and completion until retention drops it, and any move to
// blocked or discarded clears it.
Snapshot json.RawMessage
}
Decision is what admission wrote onto a record with its verdict. The pointer fields stay intake's; these are admission's, and dispatch starts a task from them.
type DecisionAction ¶
type DecisionAction string
DecisionAction is a person's decision on a record, in the words the decisions table records it with.
const ( // DecisionRedispatch authorized the record to run. DecisionRedispatch DecisionAction = "redispatch" // DecisionDiscard closed it without running it. DecisionDiscard DecisionAction = "discard" )
type Delivery ¶
type Delivery string
Delivery is an event's delivery state on a task. It moves forward only.
const ( // DeliveryAdmitted is on the task and not yet handed to a worker. DeliveryAdmitted Delivery = "admitted" // DeliveryExposed was handed to a worker, which may have acted on it. DeliveryExposed Delivery = "exposed" // DeliveryDelivered was acknowledged by the worker. DeliveryDelivered Delivery = "delivered" // DeliveryCompleted has the worker's outcome. DeliveryCompleted Delivery = "completed" )
type Destination ¶
type Destination struct {
BucketID int64
Kind MessageKind
RecordingID int64
}
Destination is where a lifecycle message goes.
type DiscardResult ¶
type DiscardResult struct {
EventID int64 `json:"event_id"`
FromState RecordState `json:"from_state"`
FromReason string `json:"from_reason,omitempty"`
FromOutcome Outcome `json:"from_outcome,omitempty"`
// Already says the record was discarded by a person before; nothing
// changed.
Already bool `json:"already_discarded,omitempty"`
// Canceled counts lifecycle messages still pending for the event that
// will not be sent.
Canceled int `json:"canceled_messages"`
}
DiscardResult is what a discard did.
type DispatchStatus ¶
type DispatchStatus struct {
TaskID int64 `json:"task_id"`
AttemptID string `json:"attempt_id"`
State string `json:"state"`
StopReason string `json:"stop_reason,omitempty"`
LaunchedAt time.Time `json:"launched_at"`
EndedAt *time.Time `json:"ended_at,omitempty"`
Events []DispatchedEvent `json:"events"`
}
DispatchStatus is one attempt and the outcomes of the events on its task.
type DispatchedEvent ¶
type DispatchedEvent struct {
EventID int64 `json:"event_id"`
Delivery string `json:"delivery"`
Outcome string `json:"outcome,omitempty"`
ReplyID *int64 `json:"reply_id,omitempty"`
// Withdrawn is an exposure taken back after a start that ran nothing.
Withdrawn bool `json:"withdrawn,omitempty"`
}
DispatchedEvent is one event's delivery and outcome on a task.
type EntryClass ¶
type EntryClass string
EntryClass is how the feed re-entered after a 410, decided by the cursor the server served — never by what the connector would have chosen.
const ( // EntryPresent is a resume at since=now: the entry takes the head and // history below it is gone. EntryPresent EntryClass = "present" // EntryReplay is a resume at since=<id>: the entry is a replay that // checkpoints from its first poll page. EntryReplay EntryClass = "replay" // EntryUnknown is a resume whose cursor could not be read. EntryUnknown EntryClass = "unknown" )
type Gap ¶
type Gap struct {
ID int64
DetectedAt time.Time
Class GapClass
// EpochAfterID is set exactly when Class is GapEpoch. It is a pointer
// because the distinction between "absent" and "zero" is the whole of the
// difference between the two 410s.
EpochAfterID *int64
EntryClass EntryClass
Note string
}
Gap is one recorded 410.
type GapClass ¶
type GapClass string
GapClass distinguishes the feed's two 410s. They mean different things and must never share a recovery path.
const ( // GapEpoch is the account feed's 410: the held position fell below the // feed's epoch, and epoch_after_id names where servable history begins. // The served resume URL is followed as given. GapEpoch GapClass = "epoch" // GapRetention is the inbox lane's 410: the held position fell out of the // 30-day retention window, and there is no epoch — epoch_after_id is // absent, not zero. The account lane must never produce one, and a // connector that sees one has been told something it does not model. GapRetention GapClass = "retention" )
type GapStatus ¶
type GapStatus struct {
ID int64 `json:"id"`
DetectedAt time.Time `json:"detected_at"`
Class string `json:"class"`
EpochAfterID *int64 `json:"epoch_after_id,omitempty"`
EntryClass string `json:"entry_class,omitempty"`
}
GapStatus is one recorded 410.
type HandoffLine ¶
type HandoffLine struct {
Type string `json:"type"` // "request"
EventID int64 `json:"event_id"`
EventType string `json:"event_type"`
Trigger string `json:"trigger"`
Recording HandoffRecording `json:"recording"`
ReplyTo InstructionReply `json:"reply_to"`
RequesterID int64 `json:"requester_id"`
// RequesterName is known when the requester wrote the recording.
RequesterName string `json:"requester_name,omitempty"`
// Acknowledge says a person asked for something. A comment on a thread
// the agent follows, or a completion, is context and is not acknowledged.
Acknowledge bool `json:"acknowledge"`
Content string `json:"content"`
ContentUpdatedAt time.Time `json:"content_updated_at"`
}
HandoffLine is one trusted request, as the reader gets it. It is the instruction a worker used to pull, written out: the agent's own mention stripped, nothing more than the record carries.
type HandoffOptions ¶
type HandoffOptions struct {
Ledger *Ledger
// Served is connect.json's served projects as they are now.
Served func() (map[int64]admission.Project, error)
// Buckets narrows the served projects to a run's --project, when given.
Buckets []int64
// AgentID is the agent's Person id, whose mentions are stripped.
AgentID int64
Lines *ndjson.Writer
Logger *slog.Logger
// Started is when the run started. Zero means now.
Started time.Time
// Interval is how often admitted records are looked for. Zero means
// DefaultHandoffInterval.
Interval time.Duration
}
HandoffOptions configure RunHandoff.
type HandoffRecording ¶
type HandoffRecording struct {
BucketID int64 `json:"bucket_id"`
ProjectName string `json:"project_name,omitempty"`
RecordingID int64 `json:"recording_id"`
Type string `json:"type"`
Title string `json:"title"`
URL string `json:"url"`
}
HandoffRecording points at the recording the request is about.
type HeldStatus ¶
type HeldStatus struct {
EventID int64 `json:"event_id"`
EventType string `json:"event_type"`
Trigger string `json:"trigger,omitempty"`
BucketID int64 `json:"bucket_id"`
RecordingURL string `json:"recording_url,omitempty"`
Reason string `json:"reason,omitempty"`
Generation int64 `json:"generation"`
UpdatedAt time.Time `json:"updated_at"`
}
HeldStatus is a held record.
type Hold ¶
type Hold struct {
// Generation is the intake generation the latest hold opened. Records of
// earlier generations were tagged for review.
Generation int64 `json:"generation"`
Cause HoldCause `json:"cause"`
HeldBy string `json:"held_by"`
HeldAt time.Time `json:"held_at"`
}
Hold is the standing hold marker.
type HoldResult ¶
type HoldResult struct {
Hold Hold `json:"hold"`
// Tagged is how many non-terminal records were tagged for review.
Tagged int `json:"tagged_for_review"`
// Held is how many of them were waiting for a worker and are now held.
Held int `json:"held"`
}
HoldResult is what setting a hold did.
type HoldStatus ¶
type HoldStatus struct {
Generation int64 `json:"generation"`
Cause string `json:"cause"`
HeldBy string `json:"held_by"`
HeldAt time.Time `json:"held_at"`
}
HoldStatus is the hold marker.
type Hooks ¶
type Hooks struct {
// VerdictCommitted runs in admission's verdict transaction, after the
// verdict is written: where the guard acknowledgement and the holding
// reply are called for.
VerdictCommitted func(ctx context.Context, tx Tx, v CommittedVerdict) error
// TaskLaunched runs in LaunchTask's transaction.
TaskLaunched func(ctx context.Context, tx Tx, launch Launch) error
// AttemptEnded runs in EndAttempt's transaction, after every event is
// settled: where the attempt's completion message is called for.
AttemptEnded func(ctx context.Context, tx Tx, s Settlement) error
// StillRunning runs in StillRunning's transaction.
StillRunning func(ctx context.Context, tx Tx, tick StillRunningTick) error
// RecordDecided runs in the transaction of a person's decision on a
// record — a redispatch or a discard: where a notice already posted stops
// asking for anything, and is retracted.
RecordDecided func(ctx context.Context, tx Tx, d RecordDecision) error
}
Hooks run inside the transactions of the ledger's lifecycle transitions. A hook's error rolls the transition back. Set them once, before the ledger is used.
type ImportResult ¶
type ImportResult struct {
// Tombstoned counts records closed as discarded(imported_done), Inserted
// the tombstones written for events the ledger had never seen.
Tombstoned int `json:"tombstoned"`
Inserted int `json:"tombstones_inserted"`
// AlreadyTerminal counts done entries whose record had already finished.
AlreadyTerminal int `json:"already_terminal"`
// Tagged counts non-terminal records tagged for review, and Held those
// of them that were waiting for a worker and are now held.
Tagged int `json:"tagged_for_review"`
Held int `json:"held"`
}
ImportResult is what an import did.
type InstanceHolderInfo ¶
InstanceHolderInfo is what a running connector wrote beside its lock.
func InstanceHolder ¶
func InstanceHolder(dir, accountID string, agentPersonID int64) (InstanceHolderInfo, bool)
InstanceHolder reads what a connector holding the lock in dir wrote about itself, without taking the lock: status must not make a starting connector find its own lock held. It is diagnostic, and can be stale after a crash.
type InstanceLock ¶
type InstanceLock struct {
// contains filtered or unexported fields
}
InstanceLock is the refusal of a second connector on one agent identity.
It is keyed on the ACCOUNT and the agent's Person id, not on the profile name. Two profiles can hold credentials for the same agent, and the failure this prevents is not "the same configuration twice" — it is one agent's mentions being dispatched twice, which is a property of the identity, not of the file that names it.
The kernel drops a flock when the holding descriptor closes, process death included, so a crashed connector cannot wedge the lock and there is no stale-lock reaping to get wrong. The metadata written beside it is diagnostic only: the lock is the lock.
func AcquireInstanceLock ¶
func AcquireInstanceLock(dir, accountID string, agentPersonID int64, now time.Time) (*InstanceLock, error)
AcquireInstanceLock takes the lock for one account and agent, or refuses.
func (*InstanceLock) Path ¶
func (l *InstanceLock) Path() string
Path is the lock file, for diagnostics.
type Instruction ¶
type Instruction struct {
EventID int64 `json:"event_id"`
EventType string `json:"event_type"`
Trigger string `json:"trigger"`
Class string `json:"class,omitempty"`
Recording InstructionRecording `json:"recording"`
ReplyTo InstructionReply `json:"reply_to"`
RequesterID int64 `json:"requester_id"`
// Acknowledge says a person asked for something: the worker acknowledges
// it and reports the id through ack_dispatch.
Acknowledge bool `json:"acknowledge"`
// GuardAcknowledged says the connector's guard already acknowledged the
// event, so the worker does not acknowledge it again.
GuardAcknowledged bool `json:"guard_acknowledged"`
Delivery Delivery `json:"delivery"`
// Content is the instruction as admission read it, with the agent's own
// mention stripped. The live recording may be newer.
Content string `json:"content"`
ContentUpdatedAt time.Time `json:"content_updated_at"`
}
Instruction is what get_dispatch hands a worker. It is an allowlist: every field is named here, and nothing the ledger holds reaches a worker unless it is one of them. No directory, no feed position, no token.
type InstructionRecording ¶
type InstructionRecording struct {
BucketID int64 `json:"bucket_id"`
RecordingID int64 `json:"recording_id"`
Type string `json:"type"`
Title string `json:"title"`
URL string `json:"url"`
}
InstructionRecording points at the recording the event is about.
type InstructionReply ¶
InstructionReply is where the worker's acknowledgement and reply go.
type Intake ¶
type Intake struct {
// contains filtered or unexported fields
}
Intake is the feed's delivery path: write the pointer, hand over the id.
Everything else — reading the recording, judging it, dispatching it — is downstream of the queue, so a slow admission or a busy dispatcher is absorbed by the backlog rather than felt by the socket — until the backlog reaches its pause threshold, where intake stops reading the feed on purpose rather than let the queue grow without bound.
func (*Intake) CheckpointKey ¶
func (in *Intake) CheckpointKey() eventfeed.CheckpointKey
CheckpointKey is the identity this intake's position is stored under.
func (*Intake) Run ¶
Run consumes the feed until ctx is canceled or the feed terminates.
It reconnects on its own for two reasons only — membership, and a stored position refused before this run had a safe re-entry of its own. Everything else the feed can recover from, it recovers from inside the package.
type Intent ¶
type Intent struct {
ID int64
Key string
Kind IntentKind
State IntentState
// EventID is the event a guard or holding reply answers for; zero for a
// per-attempt intent.
EventID int64
// TaskID and AttemptID are set on per-attempt intents.
TaskID int64
AttemptID string
Occurrence int
// Retracts is the intent whose posted message a retraction answers; zero
// on every other kind.
Retracts int64
Destination Destination
// Body is the message exactly as it is posted, rendered from records when
// the intent was written.
Body string
CreatedAt time.Time
NotBefore time.Time
SendingAt *time.Time
FinishedAt *time.Time
ReceiptID *int64
// Note says why an intent is canceled or indeterminate.
Note string
// ResolvedBy names the person who resolved an indeterminate intent.
ResolvedBy string
// ReconcileFailures counts listings that failed for a sending intent;
// ReconcileAt is when the next is due, nil when none failed.
ReconcileFailures int
ReconcileAt *time.Time
}
Intent is one lifecycle message.
type IntentFilter ¶
type IntentFilter struct {
States []IntentState
Kinds []IntentKind
EventID int64
// Limit is the most returned, newest first; zero for all.
Limit int
}
IntentFilter selects intents. Zero values select everything.
type IntentKind ¶
type IntentKind string
IntentKind is what a lifecycle message answers for.
const ( // IntentGuardAck is the fixed-form acknowledgement a guard posts when no // worker called get_dispatch in time. One per event. IntentGuardAck IntentKind = "guard_ack" // IntentHoldingReply answers a request the connector is not going to // start work on until a person changes something: a mention or // assignment in a project the connector does not serve. One per event. IntentHoldingReply IntentKind = "holding_reply" // IntentStillRunning is one still-running notice. One per attempt and // occurrence. IntentStillRunning IntentKind = "still_running" // IntentCompletion is an attempt's completion notice. One per attempt. IntentCompletion IntentKind = "completion" // IntentRetraction answers an ask a posted notice made: it says the thing // the notice asked a person to do has been done, or been decided against. // One per posted message and the event whose ask it answers. IntentRetraction IntentKind = "retraction" )
type IntentResolution ¶
type IntentResolution struct {
Resolution Resolution
// ReceiptID is the message a ResolveSent names.
ReceiptID int64
// By names who decided, for the record. Required.
By string
}
IntentResolution is a person's decision and who made it.
type IntentState ¶
type IntentState string
IntentState is where an intent is.
const ( // IntentPending is written and not yet asked for. IntentPending IntentState = "pending" // IntentSending was claimed for a request; the request may or may not // have reached Basecamp. IntentSending IntentState = "sending" // IntentSent has its receipt. IntentSent IntentState = "sent" // IntentIndeterminate could not be reconciled unambiguously. It is never // sent again automatically; a person decides. IntentIndeterminate IntentState = "indeterminate" // IntentCanceled was never sent: nothing called for it any more (a guard // get_dispatch canceled), or Basecamp refused the request, which creates // nothing. IntentCanceled IntentState = "canceled" // IntentAbandoned is an indeterminate intent a person decided not to // send. IntentAbandoned IntentState = "abandoned" )
type IntentStatus ¶
type IntentStatus struct {
ID int64 `json:"id"`
Kind string `json:"kind"`
EventID int64 `json:"event_id,omitempty"`
AttemptID string `json:"attempt_id,omitempty"`
BucketID int64 `json:"bucket_id"`
MessageKind string `json:"message_kind"`
RecordingID int64 `json:"recording_id"`
SendingAt *time.Time `json:"sending_at,omitempty"`
Note string `json:"note,omitempty"`
}
IntentStatus is a lifecycle message waiting for a person. The body is not shown.
type Lane ¶
type Lane string
Lane names which lane first served an event. It is diagnostic: dedupe is by id, and the same event ordinarily arrives on both.
func LaneOf ¶
LaneOf says which lane served an event.
The rows are not labeled, but the two shapes differ: actor_type and visible_to_clients are push-lane transport fields that poll rows omit, and the SDK keeps both presence-bearing — a string whose empty value is outside the vocabulary, and a *bool — rather than defaulting them, precisely so this is answerable.
It labels records and pointer lines. It decides nothing about re-entry: the last poll-served id is taken from what the poll source served, not from guessing which lane a delivery came from.
type Launch ¶
type Launch struct {
TaskID int64
// Token binds the worker to the task. It is returned once and stored
// only as a hash.
Token string
AttemptID string
// EventIDs are the task's events, originating first. Only the originating
// event is exposed; the rest wait at delivery admitted.
EventIDs []int64
ConversationKey string
Driver string
LaunchedAt time.Time
// DeadlineAt is zero when the task has no deadline.
DeadlineAt time.Time
}
Launch is a task written launching.
type LaunchSpec ¶
type LaunchSpec struct {
// EventID is the originating event: an admitted or queued record.
EventID int64
// Served is the served projects the caller decided this launch from,
// already cut to the run's --project scope. For the dispatcher it is the
// one reading of connect.json that this pass took, handed down — not a
// fresh read, and not a promise about what the file says at the moment
// the transaction commits. Records on the originating record's
// conversation join its task only from its own project, and only while
// that project is among these.
Served []int64
// Driver is the driver's name.
Driver string
// Deadline is how long the task may run; zero for none.
Deadline time.Duration
}
LaunchSpec asks for a task and its first attempt.
type Ledger ¶
type Ledger struct {
// contains filtered or unexported fields
}
Ledger is the connector's durable memory: the dedupe authority, the feed position, and the record of what was lost.
It is a SQLite file rather than a set in memory because every promise the connector makes about a crash rests on the answer to "have I seen this id before?" surviving the crash.
func OpenExistingLedger ¶
OpenExistingLedger opens a ledger the connector already created, for a process that reads and reports into it rather than owns it — a worker's MCP server. It never creates the file and never migrates: a different basecamp binary started as a worker must not change the schema under the connector that holds it, so a ledger at any other schema version is refused.
Neither the privacy check nor SQLite may create the file on this path: the check only inspects, and the database is opened with mode=rw, so a ledger removed at any moment is an error rather than a new empty one.
func OpenLedger ¶
OpenLedger opens (creating if absent) the ledger at path and brings its schema up to date.
func OpenLedgerReadOnly ¶
OpenLedgerReadOnly opens an existing ledger for reading only: no migration, no write to the database, no lock. status uses it beside a running connector (invariant 8). SQLite may create the WAL sidecars of a cleanly closed ledger to read it; they sit in the ledger's private directory.
The file must already exist, and it is refused unless it is private, as OpenLedger refuses it. A ledger an older binary wrote, which the running connector has not yet migrated, is refused: its columns are not the ones this build reads.
func (*Ledger) AdoptReply ¶
AdoptReply links a reply to an event whose outcome is unknown. The outcome stays unknown (invariant 6).
func (*Ledger) AdoptionCandidates ¶
AdoptionCandidates lists a settled task's events a reply could be adopted for.
func (*Ledger) AuthorizedBlocked ¶
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 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 re-run on a timer could only repeat the same answer for.
func (*Ledger) Close ¶
Close releases the ledger's handle. A second Close is harmless: the file is released once, so a defensive extra call cannot take the entry away from another Ledger still holding the same file.
func (*Ledger) CloseLoss ¶
CloseLoss ends a loss's reconciliation: everything still missing becomes unrecovered, and the loss is resolved. It reports how many ids were left unrecovered.
func (*Ledger) CountInState ¶
CountInState counts the records in state.
func (*Ledger) CreateTask ¶
CreateTask puts records on a new task at delivery admitted, with the acknowledgement guard armed for the records whose verdict asks for one, and moves each record to dispatched in the same transaction: a record is dispatched exactly while a live task carries it.
An event is on at most one live task. A second task for an event whose task was not superseded is refused with ErrEventOnLiveTask, and nothing is written. The refusal is the database's own (task_events_one_live_task), so a retried or concurrent launch cannot get past it. A redispatch supersedes the old task first.
func (*Ledger) Discard ¶
Discard closes a held, blocked or unknown record without running it, as discarded(by_operator), and records who decided. Anything else is refused with ErrDecisionRefused.
func (*Ledger) Dispatch ¶
Dispatch binds the ledger to a worker's task token, refusing one that names no live task now. The token is checked again on every call, in the call's own transaction, so a redispatch that supersedes it later takes effect at once. agentID is the agent's Person id, whose own mentions are stripped from the instructions handed out.
func (*Ledger) DropContent ¶
func (l *Ledger) DropContent(ctx context.Context, discardedBefore, completedBefore time.Time) (int, error)
DropContent drops the pointer payload of terminal records past their retention window and leaves the tombstone: id, state, outcome timestamps.
The tombstone is what makes an explicit replay safe forever, so it is never deleted — only the payload goes. Non-terminal records are never touched: their payload is the only copy of what intake was told. Nor is a completed record a person redispatched while its task was live: that task's end admits it, and admitted needs its snapshot.
func (*Ledger) DueBlockedRetries ¶
DueBlockedRetries returns one page of the blocked records whose next retry has come, in scope, the longest overdue first, at most Limit of them and all sorting after the cursor.
The schedule is not re-derived here. next_retry_at is what admission.NextBlockedRetry answered when the verdict was written, so this is an indexed comparison against a stored moment rather than a scan that dates every blocked row the ledger has ever held. A record the schedule is finished with has no next_retry_at and is not read at all (Copilot on #770).
That only holds if the query is ordered the way events_next_retry is. Ordered by id instead, SQLite planned it against events_state_id and walked every blocked row the ledger has ever held, filtering next_retry_at after the fact — the index present, the schema right, the tests green, and the scan the stored due time exists to remove happening anyway (Copilot on #770). The plan is the thing that has to be checked, and TestTheSweepsQueriesUseTheDueTimeIndex checks it.
func (*Ledger) EndAttempt ¶
func (l *Ledger) EndAttempt(ctx context.Context, end AttemptEnd) (Settlement, error)
EndAttempt ends a live attempt with its stop reason, supersedes the task's token, settles every event on the task, and ends the task, in one transaction (invariants 3 to 5). Ending an attempt that already ended is ErrNoLiveAttempt.
func (*Ledger) ExposeEvent ¶
ExposeEvent writes a follow-up exposed by the live attempt, and moves its record to dispatched, before a prompt about it is sent (invariant 1). It reports false when the event was already exposed — by get_dispatch, say — which is not an error.
func (*Ledger) ExtendLossDeadline ¶
ExtendLossDeadline pushes a loss's window out, for time the walk spent waiting on the server rather than working.
func (*Ledger) ForgetPosition ¶
ForgetPosition drops the held position for key while keeping the last poll-served id — the 409 path. The server refused the position because it was minted for a different filter set; the id the poll lane had reached is still true, and is where the new digest re-enters.
func (*Ledger) Held ¶
Held reports whether the hold marker stands. Its signature is OutboxOptions.Paused's.
func (*Ledger) HoldMarker ¶
HoldMarker reads the standing hold, if any.
func (*Ledger) Import ¶
func (l *Ledger) Import(ctx context.Context, r Reconciliation, by string) (ImportResult, error)
Import applies a reconciliation in one transaction (invariant 7): a tombstone for each entry decided done, and the review tag on every other non-terminal record, mapped or not, each keeping its state and blocking reason. An entry that cannot be applied — a done entry whose record a worker holds, a held entry for an event the ledger never saw — refuses the whole file, and nothing is written.
func (*Ledger) IsLifecycleReceipt ¶
IsLifecycleReceipt reports whether a message id is the receipt of one of the connector's own lifecycle messages of that kind.
func (*Ledger) JoinConversation ¶
func (l *Ledger) JoinConversation(ctx context.Context, taskID int64, served []int64) ([]int64, error)
JoinConversation puts the records on a live task's conversation, in the task's own project, that wait for a worker onto the task, at delivery admitted, and returns their ids. served is the projects connect.json serves now, already cut to the run's --project scope; a task whose project is not among them takes nothing.
A task that has ended takes none: they start a task of their own. Nor does a task a redispatch superseded while it runs: its worker's token is refused, so what joined it could only end unknown. Nor does any task while the hold marker stands: joining is a hand-off to a worker (ledger_hold.go).
func (*Ledger) LastPollServedID ¶
LastPollServedID returns the last id the poll lane served under key, or zero if it has served none.
func (*Ledger) LaunchTask ¶
LaunchTask writes a task, its first attempt as launching, and its originating event exposed, in one transaction (invariant 1). Records on the same conversation that wait for a worker join the task at delivery admitted.
func (*Ledger) LineageHasPosition ¶
LineageHasPosition reports whether this consumer holds a position under any filter set.
func (*Ledger) LineagePollServedID ¶
func (l *Ledger) LineagePollServedID(ctx context.Context, key eventfeed.CheckpointKey) (int64, error)
LineagePollServedID returns the highest id the poll lane has served to this consumer under ANY filter set: same origin, account and namespace.
It is the re-entry for a filter change. The new digest holds no position of its own, and entering it at the present would skip everything committed since the old digest's walk stopped. Events a narrower old filter excluded before this id are not recovered by widening — the spec says so, and so does this comment — but nothing after it is skipped.
func (*Ledger) LiveAttempts ¶
func (l *Ledger) LiveAttempts(ctx context.Context) ([]LiveAttempt, error)
LiveAttempts lists every attempt not ended, oldest first. On start they are all a previous process's: launching is read as running, because the worker may exist.
func (*Ledger) Load ¶
Load returns the stored position for key. It runs once, before the first mint, and a store failure is a failure to start rather than a silent entry at the present: re-entering at the head after a crash drops everything the feed committed while we were down.
func (*Ledger) MarkRunning ¶
MarkRunning moves a launching attempt to running with its process and session.
func (*Ledger) MarkTakerUnaccounted ¶
MarkTakerUnaccounted records that the attempt's task token was delivered and the process holding it cannot be accounted for. It is the ledger's half of the same rule the socket holds in memory: a restart must not read an attempt with no taker recorded as an attempt whose token nobody took, and settle around a process that may still have it.
func (*Ledger) MarkUnrecoveredThrough ¶
MarkUnrecoveredThrough gives up on a loss's missing ids at or below id — the ones a 410's epoch has fenced off — and leaves the rest missing.
func (*Ledger) MissingIDs ¶
MissingIDs returns the ids of a loss still in state, ascending.
func (*Ledger) NoteConnection ¶
NoteConnection records the running connector's connection state, for status. detail is a short diagnostic phrase and never carries a position, a ticket or content.
func (*Ledger) NotePollServed ¶
func (l *Ledger) NotePollServed(ctx context.Context, key eventfeed.CheckpointKey, eventID int64) error
NotePollServed advances the last poll-served event id for key.
This is tracked apart from the position, and apart from the last id delivered, because all three answer different questions. The position is an opaque signed token, so nothing can be compared to it or derived from it. The last delivered id is usually a live id, thirty-odd seconds ahead of the poll lane. Only the last poll-served id is a safe place to re-enter the feed after a 409 or a rejected position — anything ahead of it skips events the poll lane had not served yet.
It only ever moves forward.
func (*Ledger) OpenLosses ¶
OpenLosses returns the losses whose reconciliation has not finished, oldest first. Reconciliation resumes from this on every start.
func (*Ledger) RecordGap ¶
RecordGap writes one 410. Nothing is posted to Basecamp for a gap; it is a fact about the feed, shown by status and doctor.
func (*Ledger) RecordLoss ¶
func (l *Ledger) RecordLoss(ctx context.Context, droppedIDs []int64, now time.Time, window time.Duration, filters eventfeed.Filters) (Loss, error)
RecordLoss writes an accepted buffer overflow, in one transaction, before the handler returns Accept.
Accept means the consumer owns the incompleteness. Owning it starts with it being on disk: a crash a millisecond later must still find the loss and resume its reconciliation, and a loss that only ever lived in memory is one the next start would never know to repair. The corollary is that a failed write is a reason to refuse the signal, not to accept it anyway.
Dropped ids the ledger already holds were lost from the buffer, not from the connector; they are recorded as never_lost so the repair walk does not go looking for events it already has.
func (*Ledger) RecordProgress ¶
RecordProgress stamps the live attempt's last progress, which still-running reads.
func (*Ledger) RecordRefusal ¶
RecordRefusal records one refusal on a live attempt, at the moment the driver made or observed it (driver's "Refusals"). An attempt that has ended is ErrNoLiveAttempt: its count was settled with it.
func (*Ledger) RecordSeen ¶
RecordSeen writes the pointer as seen if the id is new, and reports whether it was.
This is the dedupe, and it is durable on purpose. The poll lane runs about thirty seconds behind the live lane, so the ordinary case is the same event arriving twice; a restart in between would let an in-memory set answer "new" to the second copy and dispatch it a second time.
A known id is never rewritten. That is what makes a terminal record a tombstone: the ledger keeps (id, state, timestamps) indefinitely even after its pointer payload is dropped, so an explicit replay of old history can never turn a finished event back into a new task.
It also resolves the id against any loss, open or closed: an event the repair walk — or the ordinary poll lane, later — serves is one the overflow did not cost us.
func (*Ledger) RecordTaker ¶
RecordTaker records the process that took the attempt's task token — the worker's MCP server, which an agent may have started in a process group of its own. A restart ends it by this record, as it ends the worker by the worker's.
func (*Ledger) RecordsInState ¶
func (l *Ledger) RecordsInState(ctx context.Context, state RecordState, limit int) ([]Record, error)
RecordsInState returns up to limit records in state, oldest event first. Every non-terminal state is re-run on start, and this is how they are found.
func (*Ledger) RecordsInStateAfter ¶
func (l *Ledger) RecordsInStateAfter(ctx context.Context, state RecordState, afterID int64, limit int) ([]Record, error)
RecordsInStateAfter returns up to limit records in state whose id is above afterID, oldest first — the paging form of RecordsInState.
func (*Ledger) Redispatch ¶
func (l *Ledger) Redispatch(ctx context.Context, eventID int64, by string, served []int64) (RedispatchResult, error)
Redispatch authorizes a record to run again, or for the first time, and records who authorized it (invariants 4 to 6).
- 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 not serve.
- held with its snapshot, in a served project and with no blocking reason: admitted.
- blocked, or held over a blocking reason: authorized as blocked, and Rerun asks the caller to run what blocked it.
- succeeded, discarded, and anything live (seen, admitted, queued, dispatched) are refused with ErrDecisionRefused.
func (*Ledger) Release ¶
Release clears the hold marker. Held records stay held; records a person authorized and records of the newest generation dispatch.
func (*Ledger) ResolveIntent ¶
ResolveIntent applies a person's decision to an indeterminate intent.
func (*Ledger) Save ¶
Save durably records position under key.
The feed calls this only for a poll page, and only once that page's events have been accepted by the consumer — which, here, means they are already rows in the events table. A live event id never reaches this method, and must not: a live id runs ahead of the poll lane's safety delay, so a position taken from one would skip everything inside that window.
func (*Ledger) SaveRepairCursor ¶
SaveRepairCursor records where the repair walk has reached.
This is the walk's own cursor and it is stored on the loss, never on the feed's checkpoint. The walk is seeded from a live id that ran far ahead of the poll lane; a checkpoint taken from it would skip every event still inside the safety delay behind it.
func (*Ledger) SchemaVersion ¶
SchemaVersion reports the highest applied migration.
func (*Ledger) SetHold ¶
SetHold sets the durable hold marker, opens a new intake generation, and tags every non-terminal record of the generations before it for review, in one transaction (invariant 3). A hold already standing keeps its first setter and time; the new generation and tags are written again.
func (*Ledger) SetRepairEntry ¶
SetRepairEntry moves a loss's walk entry to since and drops its cursor — after a 410 has fenced off everything at or below since.
func (*Ledger) SetState ¶
SetState moves a record to state with a reason, which must be empty for every state but blocked and discarded — those two are the only ones a reason explains.
The move is refused unless the lifecycle has that edge, and it is refused by the UPDATE itself rather than by a read before it: a check and a write in two statements is a race, and this is the guarantee that a finished record stays finished.
func (*Ledger) StartableRecords ¶
StartableRecords returns up to limit records waiting for a worker, the oldest per conversation, oldest first. While the hold marker stands there are none: the database would refuse their launch (ledger_hold.go).
func (*Ledger) StartableRecordsWhere ¶
StartableRecordsWhere is StartableRecords narrowed by f.
func (*Ledger) Status ¶
Status reads everything the ledger's own status shows in one read transaction, so the numbers agree with each other.
func (*Ledger) StillRunning ¶
StillRunning counts one more still-running occurrence for a live attempt, running the StillRunning hook in the same transaction.
func (*Ledger) StillScheduledBlockedIDs ¶
func (l *Ledger) StillScheduledBlockedIDs(ctx context.Context, scope BlockedRetryScope, ids []int64) ([]int64, error)
StillScheduledBlockedIDs is the subset of ids the retry schedule still has something to do about, due now or later.
It answers the only question the sweep asks — "of the records I hold a claim for, which can still be offered?" — rather than the one it used to ask, which was for the entire live backlog. For the claims held those have the same answer and wildly different costs: after an outage the schedule holds every record the outage blocked, and a sweep reading all of them once a minute to prune a handful of claims does work proportional to the backlog instead of to the claims, and delays the stranded and loss repair sharing its goroutine (Copilot on #770).
A claim is worth keeping only while the record it names can still be offered, and a record that has been decided, or whose window has passed, can never be. Asking about exactly the claimed ids is what makes pruning safe: the caller drops only the ids it asked about and did not get back, so an id this was never asked about cannot be pruned by its absence.
It reads in chunks and returns their union. A chunk dropped or shortened would read as "no longer scheduled" and prune live claims, which is the offer-twice the claim exists to prevent, so a failed chunk fails the call.
func (*Ledger) StrandedRecords ¶
StrandedRecords counts the records waiting for a worker in a project connect.json no longer serves: work admitted while the project was served, which nothing will start until a person serves it again or discards the record. buckets is the run's --project scope: work in a project this run does not hear is another run's to dispatch, not stranded, so it is not counted.
func (*Ledger) SupersedeTask ¶
SupersedeTask retires a task: its token is refused from then on, and its events are free to join a new task. An event this task never exposed returns to admitted, to be dispatched again once its conversation is free — unless something else still holds it dispatched, which invariant 4 decides and this call does not override: an earlier task's exposure that was never withdrawn keeps the record dispatched, waiting for its outcome, and a record a person or a later verdict has moved stays where it was put. An event a worker was handed stays dispatched for the same reason: that worker may have acted on it, and it waits for its settlement or a redispatch.
func (*Ledger) UnexposedEvents ¶
UnexposedEvents are the events on a task still at delivery admitted, oldest first: the follow-ups a live session has not been prompted with.
func (*Ledger) UnfinishedFromOthers ¶
UnfinishedFromOthers counts the records not yet finished that someone other than operatorID asked for: admitted, queued, held or dispatched, or blocked after its worker failed to start, by the person admission trusted at the gate (the performer, else the creator). A record only seen, or blocked on any other reason, is judged again under the trust the connector runs with now, so it isn't counted: one stuck request from someone else would otherwise keep dangerous mode off for good.
type LiveAttempt ¶
type LiveAttempt struct {
AttemptID string
TaskID int64
State AttemptState
Driver string
ConversationKey string
Process AttemptProcess
// Taker is the process the task token went to, where one took it. Its
// PID is zero when none did.
Taker AttemptProcess
// TakerUnaccounted is a token that went out to a process this attempt
// could not account for. Its PID is zero too, and the difference
// matters: nothing of such an attempt is released.
TakerUnaccounted bool
LaunchedAt time.Time
// DeadlineAt is zero when the task has none.
DeadlineAt time.Time
}
LiveAttempt is an attempt that has not ended.
type LiveWorker ¶
type LiveWorker struct {
AttemptID string `json:"attempt_id"`
TaskID int64 `json:"task_id"`
// PID, PGID and StartedAt are the recorded process: with the start time,
// the pid is an identity (driver.OwnsWorker).
PID int `json:"pid"`
PGID int `json:"pgid"`
StartedAt time.Time `json:"started_at"`
}
LiveWorker is an attempt's recorded worker process.
func (LiveWorker) Identity ¶
func (w LiveWorker) Identity() driver.Process
Identity is the process the record names, for the one-owner rule.
type Loss ¶
type Loss struct {
ID int64
DetectedAt time.Time
DroppedCount int
// RepairSince is the walk's entry: one below the lowest missing id.
RepairSince int64
// RepairCursor is the walk's own position, kept here and never written to
// the feed's checkpoint.
RepairCursor string
DeadlineAt time.Time
ResolvedAt *time.Time
// Filters is the filter set this loss was recorded under. The repair walk
// uses it, whatever the connector's current filters are: a loss recorded
// under one set and walked under another is repaired against a lane that
// never carried its events.
Filters eventfeed.Filters
// HasFilters distinguishes a loss recorded with no filters — the
// whole-account feed, the common case — from one written before losses
// carried them at all. The first is walked with no filters; the second
// with the connector's own, which is the best that row can say.
HasFilters bool
}
Loss is one accepted buffer overflow and the state of its reconciliation.
type LossState ¶
type LossState string
LossState is what became of one id the live buffer dropped.
const ( // LossMissing is an id the overflow dropped and nothing has served since. LossMissing LossState = "missing" // LossNeverLost is an id the ledger already held when the overflow was // signaled. It was dropped from the buffer, not from the connector. LossNeverLost LossState = "never_lost" // LossRecovered is an id a later poll — the repair walk or the ordinary // lane — served. LossRecovered LossState = "recovered" // LossUnrecovered is an id still missing when the repair window closed: a // late straggler, a recording deleted before it became poll-visible, or // history behind the epoch. It is reported, never hidden. LossUnrecovered LossState = "unrecovered" )
type LossStatus ¶
type LossStatus struct {
ID int64 `json:"id"`
DetectedAt time.Time `json:"detected_at"`
Dropped int `json:"dropped"`
Missing int `json:"missing"`
DeadlineAt time.Time `json:"deadline_at"`
}
LossStatus is an overflow still being reconciled.
type MembershipSource ¶
MembershipSource lists the buckets the agent can currently see.
type MessageKind ¶
type MessageKind string
MessageKind is the kind of Basecamp message an intent posts.
const ( // MessageBoost is a boost on Destination.RecordingID. MessageBoost MessageKind = "boost" // MessageComment is a comment on Destination.RecordingID. MessageComment MessageKind = "comment" // MessageChatLine is a line in the Campfire Destination.RecordingID. MessageChatLine MessageKind = "chat_line" )
type Options ¶
type Options struct {
// Origin is the API base URL. It is part of the checkpoint identity and
// the origin every continuation URL is validated against.
Origin string
// AccountID is the Basecamp account.
AccountID string
// ConsumerNamespace names this connector's checkpoint lineage. Two
// connectors in one account must not share one.
ConsumerNamespace string
// Filters is the feed's filter set. The checkpoint is keyed by its digest,
// so changing it re-enters under a new lineage.
Filters eventfeed.Filters
// SinceEventID, when positive, enters just after that event id whatever
// the ledger holds — the `--since` override.
SinceEventID int64
Ledger *Ledger
Queue *Queue
Minter eventfeed.TicketMinter
// PollsFor returns a fresh poll source per walk. The SDK's live source
// holds walk state (the order a continuation must follow), so the feed's
// connection and each repair walk get their own. Polls, when set instead,
// is shared by all of them — for test fakes that hold no walk state.
PollsFor func() eventfeed.PollSource
Polls eventfeed.PollSource
// Pointers receives one NDJSON line per newly seen event. Writes are
// serialized: an interleaved write tears a line and breaks the watcher
// reading it.
Pointers io.Writer
// Lines, when set, is the writer the pointer lines go through, and
// Pointers is ignored. One process stdout has one writer and one lock, so
// a caller wiring intake and admission onto the same stdout passes the
// same writer to both rather than relying on them to find each other.
Lines *ndjson.Writer
// Logger receives everything else. Pointer lines are the protocol;
// logging is not.
Logger *slog.Logger
// Membership, when set, drives the reconnect that makes a newly granted
// project visible on the live lane.
Membership MembershipSource
MembershipInterval time.Duration
RepairInterval time.Duration
RepairWindow time.Duration
Clock func() time.Time
Transport eventfeed.CableTransport
}
Options configures intake.
func LiveOptions ¶
LiveOptions binds intake to the live feed: the SDK's own seams over the generated operations, with their redirect guard, their two-410 mapping and their continuation checks. The caller fills in the rest (account, namespace, ledger, queue, filters).
Build the Live without a debug logger or request hooks in production: the SDK client logs request URLs through them, and a poll URL carries the feed position.
type Outcome ¶
type Outcome string
Outcome is what a worker reports for an event.
const OutcomeUnknown Outcome = "unknown"
OutcomeUnknown is an event that was exposed to a worker and never reported: whatever ended the attempt, the worker may have acted on it.
type Pointer ¶
type Pointer struct {
EventID int64 `json:"event_id"`
EventType string `json:"event_type"`
Kind string `json:"kind"`
Action string `json:"action"`
BucketID int64 `json:"bucket_id"`
CreatorID int64 `json:"creator_id"`
PerformedByID *int64 `json:"performed_by_id"`
RecordingID int64 `json:"recording_id"`
CreatedAt string `json:"created_at"`
Lane Lane `json:"lane"`
State string `json:"state"`
}
Pointer is the line intake writes to stdout for each newly seen event. It carries what the feed carried and nothing more: no title, no body, no URL, no names. Whoever wants those pays for a read.
type PositionStatus ¶
type PositionStatus struct {
Filters string `json:"filters"`
HasPosition bool `json:"has_position"`
LastPollServedID int64 `json:"last_poll_served_id"`
UpdatedAt time.Time `json:"updated_at"`
}
PositionStatus is one feed checkpoint: whether a position is held, never the position itself, which resumes the account's feed.
type PromoteOptions ¶
type PromoteOptions struct {
// ShadowDir is the shadow run's state directory, StateDir the normal
// one. Both must be on one filesystem: the ledger moves by rename.
ShadowDir string
StateDir string
AccountID string
AgentID int64
// By records who promoted.
By string
}
PromoteOptions names the two state directories of one account and agent.
type PromoteResult ¶
type PromoteResult struct {
// Already says an earlier promote finished: the normal ledger stands
// under its hold and there was no shadow ledger left to move.
Already bool `json:"already_promoted,omitempty"`
Hold Hold `json:"hold"`
Tagged int `json:"tagged_for_review"`
Held int `json:"held"`
// Ledger is the promoted ledger's path.
Ledger string `json:"ledger"`
}
PromoteResult is what a promote did.
func PromoteShadow ¶
func PromoteShadow(ctx context.Context, opts PromoteOptions) (PromoteResult, error)
PromoteShadow turns a shadow run's ledger into the connector's, held (invariant 7). In order:
- Both instance locks are taken, the shadow's and the connector's: no shadow writer and no connector is running.
- In the shadow ledger, in one transaction: the hold marker, a new generation, every non-terminal record tagged for review, waiting records held.
- The shadow ledger is checkpointed into its single database file.
- That file is renamed into the connector's state directory, which exposes it; the directory is synced.
A crash before 2 commits leaves the untouched shadow. After it, the ledger at either path is held, and running promote again finishes the move. The hold is committed before the rename, so no unheld ledger is ever at the normal path.
type Queue ¶
type Queue struct {
// OnWarn fires when the depth first crosses the warning threshold, and
// OnRecover when it falls back below. Both are optional.
OnWarn func(depth int)
OnRecover func(depth int)
// OnPause fires when an offer begins waiting for room, and OnResume when
// it stops. The feed is not being consumed in between.
OnPause func(depth int)
OnResume func(depth int)
// contains filtered or unexported fields
}
Queue is the seam between intake and admission: intake writes a pointer and hands over an id, admission reads it when it gets there. Two queues with visible depth rather than one pipeline, so a busy dispatcher is absorbed up to the pause threshold instead of being felt on the socket — and so the place where work is piling up is the place the depth is showing. At that threshold the offer waits: the backlog is bounded, and the pause is the backpressure reaching the feed, reported rather than hidden.
func (*Queue) Offer ¶
Offer hands an event id to admission, waiting for room when the backlog is at the pause threshold. Waiting here is the pause: the caller is the feed's delivery path, and it is not reading the feed while it waits.
func (*Queue) Paused ¶
Paused reports whether an offer is currently waiting for room — which is to say whether the feed is being consumed.
func (*Queue) SetLogger ¶
SetLogger sets where a callback that panicked is reported. Safe to call while the queue is in use.
type Receipt ¶
type Receipt struct {
EventID int64 `json:"event_id"`
Delivery Delivery `json:"delivery"`
AckID *int64 `json:"ack_id,omitempty"`
Outcome Outcome `json:"outcome,omitempty"`
ReplyID *int64 `json:"reply_id,omitempty"`
Links []string `json:"links,omitempty"`
}
Receipt is the delivery state an acknowledgement or completion left.
type Reconciliation ¶
type Reconciliation struct {
Version int `json:"version"`
Entries []ReconciliationEntry `json:"entries"`
}
Reconciliation is the cutover's reconciliation file: the old connector's handled entries, each mapped to a feed event and decided.
func ParseReconciliation ¶
func ParseReconciliation(data []byte) (Reconciliation, error)
ParseReconciliation reads a reconciliation file strictly: unknown fields, a second entry for one event, and a decision other than done or held are refused, so a file that means something else is never half-understood.
func (Reconciliation) Validate ¶
func (r Reconciliation) Validate() error
Validate refuses a reconciliation this build cannot apply: another version, an entry without an event, a decision that is neither done nor held, or one event decided twice. Import checks it too, so a caller that built the value itself meets the same rules as one that parsed a file.
type ReconciliationEntry ¶
type ReconciliationEntry struct {
EventID int64 `json:"event_id"`
Decision string `json:"decision"`
}
ReconciliationEntry is one mapped entry.
type Record ¶
type Record struct {
ID int64
State RecordState
Reason string
Lane Lane
EventType string
Kind string
Action string
BucketID int64
CreatorID int64
PerformedByID *int64
RecordingID int64
Details json.RawMessage
ActorType string
VisibleToClients *bool
CreatedAt time.Time
SeenAt time.Time
UpdatedAt time.Time
ContentDropped bool
// Revision counts every state write applied to the record since intake
// recorded it, a repeat of the state it already has included. A decision
// applies only at the revision it loaded.
Revision int64
// Decision is admission's latest verdict on the record; its zero value
// until one is written.
Decision Decision
}
Record is one row of the events table.
func (Record) EffectivePerformer ¶
EffectivePerformer is the id the performers/exclude_performers filters match: the delegating agent when the action was delegated, else the creator.
type RecordDecision ¶
type RecordDecision struct {
EventID int64
Action DecisionAction
At time.Time
}
RecordDecision is what RecordDecided is told: the record a person decided, how, and when.
type RecordState ¶
type RecordState string
RecordState is where an event sits in the ledger's lifecycle.
Intake only ever writes StateSeen. The rest of the vocabulary is declared here because the states are one lifecycle, and a store that cannot name the state a later card writes cannot recover it on start either.
const ( // StateSeen is a pointer intake wrote and nothing has judged yet. Every // seen record re-runs the gate on start. StateSeen RecordState = "seen" // StateAdmitted passed the gate and awaits dispatch. StateAdmitted RecordState = "admitted" // StateQueued waits behind another event on its conversation. 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 <id>`. StateBlocked RecordState = "blocked" // StateDispatched was handed to a worker. StateDispatched RecordState = "dispatched" // StateCompleted is terminal with an outcome. StateCompleted RecordState = "completed" // StateDiscarded is terminal with a verified verdict. StateDiscarded RecordState = "discarded" // StateHeld waits for a person. A record tagged for review by a hold // becomes held where it would have waited for a worker, and only a // person's redispatch or discard moves it on. It keeps its snapshot. StateHeld RecordState = "held" )
type RedispatchResult ¶
type RedispatchResult struct {
EventID int64 `json:"event_id"`
FromState RecordState `json:"from_state"`
FromReason string `json:"from_reason,omitempty"`
FromOutcome Outcome `json:"from_outcome,omitempty"`
// State is the record's state after the authorization.
State RecordState `json:"state"`
// Admitted says the record waits for a worker now.
Admitted bool `json:"admitted"`
// Pending says the record's task is still live: it is admitted in the
// transaction that ends that task.
Pending bool `json:"pending,omitempty"`
// Rerun says the record was authorized as blocked: the caller runs its
// prerequisite again (admission), which admits it when it succeeds.
Rerun bool `json:"rerun,omitempty"`
// SupersededTaskID is the task whose token this redispatch retired; zero
// when it was already retired.
SupersededTaskID int64 `json:"superseded_task_id,omitempty"`
// Worker is the replaced attempt's recorded process, still live in the
// ledger: the caller terminates it (driver.TerminateRecorded).
Worker *LiveWorker `json:"worker,omitempty"`
// Held says the hold marker stands: authorized, and nothing launches
// until release.
Held bool `json:"held,omitempty"`
}
RedispatchResult is what a redispatch did.
type ReleaseResult ¶
type ReleaseResult struct {
// Released is false when no hold stood.
Released bool `json:"released"`
Hold Hold `json:"hold,omitzero"`
// StillHeld counts held records, which stay held.
StillHeld int `json:"still_held"`
}
ReleaseResult is what a release did.
type Resolution ¶
type Resolution string
Resolution is a person's decision on an indeterminate intent.
const ( // ResolveSent says the message is in Basecamp: ReceiptID names it. ResolveSent Resolution = "sent" // ResolveAbandon says it is not to be sent. ResolveAbandon Resolution = "abandon" // ResolveResend authorizes sending it again: the intent returns to // pending. Only a person may choose this; nothing automatic does. ResolveResend Resolution = "resend" )
type SDKMembership ¶
type SDKMembership struct {
Client *basecamp.AccountClient
}
SDKMembership lists the buckets the agent can see, for intake's reconnect.
type SettledEvent ¶
type SettledEvent struct {
EventID int64
// Outcome is the reported outcome, or unknown for an event exposed and
// never reported. Empty for an event never exposed, or withdrawn.
Outcome Outcome
// Reported is whether the outcome is the worker's own report.
Reported bool
ReplyID *int64
// Returned is an event never exposed: it waits for a task of its own.
Returned bool
// Withdrawn is an exposure withdrawn after a start that ran nothing; the
// record is admitted again, or blocked(spawn_failed) when it already was
// once.
Withdrawn bool
// Blocked is a withdrawal refused a second automatic retry.
Blocked bool
// Decided is a record a person has already redispatched or discarded, so
// the completion notice asks nothing of them.
Decided bool
// Pulled is whether a worker asked for the event's instruction. An event
// the launch exposed and no worker pulled is one the worker never picked
// up: it stopped before it began.
Pulled bool
}
SettledEvent is one event's state after its task ended.
type Settlement ¶
type Settlement struct {
TaskID int64
AttemptID string
Stop StopReason
// SpawnFailed repeats AttemptEnd.SpawnFailed.
SpawnFailed bool
// OriginatingEventID is the task's originating event.
OriginatingEventID int64
Events []SettledEvent
}
Settlement is what ending an attempt did to its task.
type StartableFilter ¶
type StartableFilter struct {
// Served are the served project ids the caller decided this query from,
// already narrowed to --project — one reading, not a promise about the
// file at any later moment. A record in any other project is not
// startable. Empty means nothing is.
Served []int64
Limit int
}
StartableFilter narrows StartableRecordsWhere to what the dispatcher can start now, in the query itself: a record it would skip must never take a place in the window, or a backlog it cannot start starves everything behind it.
type StateDirError ¶
type StateDirError struct {
// Dir is the directory as given, made absolute.
Dir string
// Root is the connector's state root, the only place a state directory
// lives.
Root string
// Account is the account the directory names, empty when it names none;
// Want is the account it had to name.
Account, Want string
// Why is the rule it failed.
Why StateDirProblem
}
StateDirError says which rule a state directory failed, in fields a caller can build its own message from rather than by reading this one.
func (*StateDirError) Error ¶
func (e *StateDirError) Error() string
func (*StateDirError) Unwrap ¶
func (e *StateDirError) Unwrap() error
type StateDirProblem ¶
type StateDirProblem string
StateDirProblem is why a state directory was refused.
const ( // StateDirElsewhere is a directory outside the state root. StateDirElsewhere StateDirProblem = "outside the connector's state root" // StateDirMisnamed is a directory not named <account>-<agent person id>. StateDirMisnamed StateDirProblem = "not named <account>-<agent person id>" // StateDirOtherAccount is another account's state directory. StateDirOtherAccount StateDirProblem = "another account's" )
type Status ¶
type Status struct {
SchemaVersion int `json:"schema_version"`
// Connection is the last run's own record, if one ran on this build.
Connection *ConnectionStatus `json:"connection,omitempty"`
// Hold is the standing hold marker.
Hold *HoldStatus `json:"hold,omitempty"`
Generation int64 `json:"generation"`
Positions []PositionStatus `json:"positions"`
Gaps []GapStatus `json:"gaps"`
Losses []LossStatus `json:"open_losses"`
Unrecovered int `json:"unrecovered_ids"`
// Queues counts records by state; Blocked counts blocked records by reason.
Queues map[string]int `json:"queues"`
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.
Review int `json:"review_tagged"`
AuthorizedBlocked int `json:"authorized_blocked"`
RedispatchPending int `json:"redispatch_pending"`
Tasks []TaskStatus `json:"live_tasks"`
Indeterminate []IntentStatus `json:"indeterminate_intents"`
Held []HeldStatus `json:"held_records"`
Dispatches []DispatchStatus `json:"dispatches"`
}
Status is what `basecamp connect status` shows. Every field is ids, states, counts and timestamps: no content, no feed position, no token or token hash (invariant 8).
type StillRunningTick ¶
type StillRunningTick struct {
AttemptID string
TaskID int64
// Occurrence counts from 1 per attempt.
Occurrence int
// ProgressAt is the attempt's last progress; zero when none was seen.
ProgressAt time.Time
}
StillRunningTick is one still-running occurrence of a live attempt.
type StopReason ¶
type StopReason string
StopReason is why an attempt ended. It is not an outcome.
const ( // StopFinished is a clean stop: the turn ended and the worker exited 0. StopFinished StopReason = "finished" // StopFailed is a refusal, a stop the connector did not ask for, a // non-zero exit, or a worker that could not be started. StopFailed StopReason = "failed" // StopDeadline is the task's deadline. StopDeadline StopReason = "deadline" // StopShutdown is the connector shutting down. StopShutdown StopReason = "shutdown" // StopLost is a worker that went away with a turn in flight, or one a // restarted connector found. StopLost StopReason = "lost" )
type TaskDispatch ¶
type TaskDispatch struct {
// contains filtered or unexported fields
}
TaskDispatch is the ledger as one worker sees it: the events on the task its token names, and nothing else. There is no listing; a worker never reads other tasks.
func (*TaskDispatch) Ack ¶
Ack records the worker's acknowledgement: delivery moves to delivered, and ackID, when given, is the worker's own boost or comment. A repeat — a lost tool response retried — answers the same receipt.
func (*TaskDispatch) Complete ¶
func (d *TaskDispatch) Complete(ctx context.Context, eventID int64, c Completion) (Receipt, error)
Complete records the worker's outcome and acknowledges the event if it was not already. The report is recorded whatever has happened to the record since the worker was handed it, because it is what the worker did; the record moves to completed when it is dispatched. A repeat of the same report answers the same receipt; a different one is refused, because a reported outcome stands.
func (*TaskDispatch) Get ¶
func (d *TaskDispatch) Get(ctx context.Context, eventID int64) (Instruction, bool, error)
Get returns the instruction for eventID, or for the earliest event on the task not yet acknowledged when eventID is zero; ok is false when there is none. Handing out an event that was never exposed writes exposed before the instruction is returned, and cancels an armed guard. A repeat returns the same instruction and writes nothing.
Only an event whose record is still on the way to a worker is handed out: dispatched, or completed, with its content. The earliest skips any other, so one event withdrawn or blocked never hides the rest of the task.
type TaskGrant ¶
TaskGrant is a new task and the token that binds a worker to it. The token is returned once and stored only as a hash.
type TaskStatus ¶
type TaskStatus struct {
TaskID int64 `json:"task_id"`
AttemptID string `json:"attempt_id"`
State string `json:"state"`
Driver string `json:"driver"`
PID int `json:"pid,omitempty"`
PGID int `json:"pgid,omitempty"`
// ProcessStartedAt is the start time recorded with the pid: with it, the
// pid is an identity (driver.OwnsWorker).
ProcessStartedAt *time.Time `json:"process_started_at,omitempty"`
// TakerPID, TakerPGID and TakerStartedAt are the process the task token
// went to, where one took it: a worker's MCP server, which lives in a
// process group of its own.
TakerPID int `json:"taker_pid,omitempty"`
TakerPGID int `json:"taker_pgid,omitempty"`
TakerStartedAt *time.Time `json:"taker_started_at,omitempty"`
// TakerUnaccounted is the case the zero taker cannot express: the token
// was delivered and the connector could not name the process that took
// it, or the kernel stopped answering whether it is gone. The attempt is
// held for it, and a person reading this status is the one who settles
// it, so it is never shown as nothing having taken the token.
TakerUnaccounted bool `json:"taker_unaccounted,omitempty"`
// Worker and Taker are whether each recorded process is still this task's,
// as the caller established it; the ledger read leaves them empty.
Worker string `json:"worker,omitempty"`
Taker string `json:"taker,omitempty"`
LaunchedAt time.Time `json:"launched_at"`
DeadlineAt *time.Time `json:"deadline_at,omitempty"`
EventIDs []int64 `json:"event_ids"`
}
TaskStatus is a live task and its attempt.
func (TaskStatus) TakerIdentity ¶
func (t TaskStatus) TakerIdentity() driver.Process
TakerIdentity is the process the task token went to, for the same rule. It says nothing about TakerUnaccounted, which is a fact of the ledger and not a question for the kernel: a caller asks that first.
func (TaskStatus) WorkerIdentity ¶
func (t TaskStatus) WorkerIdentity() driver.Process
WorkerIdentity is the attempt's recorded worker, for the one-owner rule.
type Tx ¶
type Tx interface {
ExecContext(ctx context.Context, query string, args ...any) (sql.Result, error)
QueryRowContext(ctx context.Context, query string, args ...any) *sql.Row
QueryContext(ctx context.Context, query string, args ...any) (*sql.Rows, error)
}
Tx is a ledger transaction a hook writes in, so what the hook writes (an outbox intent) commits or rolls back with the transition that called for it.
Source Files
¶
Directories
¶
| Path | Synopsis |
|---|---|
|
Package admission turns a seen account-feed event into a verdict: admitted, queued, blocked or discarded.
|
Package admission turns a seen account-feed event into a verdict: admitted, queued, blocked or discarded. |
|
Package driver is the connector's agent boundary: how a dispatched task becomes a working coding agent, and how the connector hears what it does.
|
Package driver is the connector's agent boundary: how a dispatched task becomes a working coding agent, and how the connector hears what it does. |
|
Package ndjson writes the connector's stdout protocol: one JSON value per line, never torn.
|
Package ndjson writes the connector's stdout protocol: one JSON value per line, never torn. |
|
Package setup is `basecamp connect setup`'s half of the connector (plan step 16): connect.json — its schema, where it lives, how it is written and how it is refused — and the checks that can run before the connector does.
|
Package setup is `basecamp connect setup`'s half of the connector (plan step 16): connect.json — its schema, where it lives, how it is written and how it is refused — and the checks that can run before the connector does. |