Documentation
¶
Index ¶
- Constants
- func BackoffFor(attempts int) time.Duration
- func BootUpdateDedupeKey(sessionID string) string
- func Cancel(taskID string) error
- func Claim(agentID bson.ObjectID, serverRunning bool) (*v2.AgentTaskSchema, error)
- func Complete(taskID, leaseToken string) error
- func Enqueue(agentID, accountID bson.ObjectID, action string, data interface{}, ...) (string, error)
- func EnsureIndexes() error
- func Fail(taskID, leaseToken, errMsg string) error
- func FindByDedupeKey(agentID bson.ObjectID, dedupeKey string) (*v2.AgentTaskSchema, error)
- func FindPendingByAction(agentID bson.ObjectID, action string) (*v2.AgentTaskSchema, error)
- func Get(taskID string) (*v2.AgentTaskSchema, error)
- func HasActiveAction(agentID bson.ObjectID, action string) (bool, error)
- func InitAgentTaskService() error
- func ListForAgent(agentID bson.ObjectID, limit int64) ([]v2.AgentTaskSchema, error)
- func ReapExpiredLeases() error
- func Release(taskID, leaseToken string) error
- func RenewLease(taskID, leaseToken string) (bool, bool, error)
- func ReplacePendingPayload(agentID bson.ObjectID, action string, data interface{}) (string, error)
- func ReportProgress(taskID, leaseToken string, pct int32, msg string) error
- func Retry(taskID string) error
- func SetGate(taskID string, opts EnqueueOpts) (bool, error)
- func ShutdownAgentTaskService() error
- func StartDispatcher()
- func StopDispatcher()
- func WorkflowDedupeKey(workflowID bson.ObjectID, actionIdx int) string
- type Assignment
- type EnqueueOpts
- type Registry
Constants ¶
const (
LeaseDuration = 60 * time.Second
)
Variables ¶
This section is empty.
Functions ¶
func BackoffFor ¶
BackoffFor returns the delay before a task that has consumed `attempts` attempts may be claimed again: 5s, 10s, 20s, 40s, ... capped at 5m.
func BootUpdateDedupeKey ¶
BootUpdateDedupeKey scopes the UpdateOnStart task to one agent process, so a reconnect loop cannot queue a stack of update tasks while one is still active. It keys on the agent's session id, which survives reconnects, and not on the per-stream id.
func Cancel ¶
Cancel takes a pending task straight to terminal. A running task cannot be yanked from under the agent, so it is flagged instead; the agent picks the flag up on its next lease renewal and unwinds.
func Claim ¶
Claim atomically moves the agent's oldest due task from pending to running and mints a fencing token. It returns (nil, nil) when nothing is due.
If the agent already has a running task, `uniq_running_per_agent` rejects the write with E11000 and we report "busy" the same way as "nothing to do". Two replicas racing for the same agent therefore need no coordination: one wins, the other backs off.
func Enqueue ¶
func Enqueue(agentID, accountID bson.ObjectID, action string, data interface{}, dedupeKey string, trigger v2.TaskTrigger, opts EnqueueOpts) (string, error)
Enqueue creates a pending task. It is idempotent on dedupeKey: if an active task with the same key already exists, its id is returned instead of an error. This closes the window where a workflow step writes a task, crashes before persisting the id, and re-enqueues on restart.
func EnsureIndexes ¶
func EnsureIndexes() error
EnsureIndexes creates the indexes the queue's correctness depends on.
The two unique partial indexes are not optimizations. `uniq_running_per_agent` is what serializes tasks on an agent; `uniq_active_dedupe` is what makes an enqueue idempotent. Removing either reintroduces the double-install bug.
func Fail ¶
Fail decides between a retry and a death, and honours a pending cancellation. A cancelled task is terminal: it must never come back on the next tick.
func FindByDedupeKey ¶
FindByDedupeKey returns the newest task with this key whatever its status, or nil if there is none.
Enqueue can only adopt an *active* task, because that is all the unique index constrains. A caller that lost the id of a task which has since finished would therefore enqueue a duplicate and re-run work that already succeeded. Looking the task up by key, terminal or not, closes that window. Safe because the only keys in use are unique for all time (workflow id + action index).
func FindPendingByAction ¶
FindPendingByAction returns the agent's pending task of this action, or nil if there is none.
This exists so a caller outside the package (agentmod, re-pointing a pending startsfserver at a newer sync — see SetGate) never needs its own GetCollection("agenttasks") read: the queue owns that collection, and every query against it lives here.
func HasActiveAction ¶
HasActiveAction reports whether the agent has an ACTIVE (pending or running) task of this action.
It exists because agent-reported status is blind to task state. `agents.status.running` is still false for the whole time a startsfserver is mid-boot, so a caller that decides "is the game server running?" from that field alone decides FALSE while the game is coming up, and gates nothing. A syncmods inserted on that decision is claimable the moment the boot finishes, and rewrites the Mods directory under a live game. "Running" for gating purposes must mean "running, or about to be", and this is the second half of that.
`active` is present exactly for pending and running, and is unset on every terminal transition, so it is the whole predicate.
func InitAgentTaskService ¶
func InitAgentTaskService() error
InitAgentTaskService creates the indexes before anything can dispatch. If the indexes are missing, the queue's invariants are not enforced, so a failure here must stop the process rather than degrade silently.
func ListForAgent ¶
func ReapExpiredLeases ¶
func ReapExpiredLeases() error
ReapExpiredLeases returns abandoned tasks to the queue, then releases tasks gated behind a parent that does not exist.
The attempt was already spent at claim time, so a crash-looping agent is bounded by MaxAttempts rather than retrying forever.
func Release ¶
Release returns a task to the queue without spending an attempt. It is the graceful-shutdown path: the agent chose to stop, so it should not be punished.
func RenewLease ¶
RenewLease extends the lease and doubles as the cancellation channel. ok=false means the agent has lost the task and must abandon it.
func ReplacePendingPayload ¶
ReplacePendingPayload overwrites the data of an already-pending task of the given action, returning its id, or "" if there was none. It marshals the payload the same way Enqueue does.
This is the only safe way to update a not-yet-claimed task's payload: it matches status=pending only, never running, because a running task's payload is already in the agent's hands and rewriting it here would not reach the agent. Callers that need "replace if pending, otherwise enqueue" must not fall back to Enqueue's dedupe-adoption on failure to find a pending task, since adoption also matches a RUNNING task and would ship it a payload update it can never see.
func Retry ¶
Retry resurrects a dead task with a fresh attempt budget. If an identical action is already active, uniq_active_dedupe rejects the write and the caller gets a usable message rather than a second install.
func SetGate ¶
func SetGate(taskID string, opts EnqueueOpts) (bool, error)
SetGate makes a PENDING task's claim gates exactly the ones in opts, replacing whatever it carried before. It is a no-op on any other status: a running task's gates have already been evaluated.
It exists because Enqueue's dedupe-adoption path ignores opts — an adopted task keeps the gating it was created with — and because a caller's plan may need to move an existing task onto a different gate entirely (escalating a deferred syncmods into a stop -> sync chain must CLEAR requiresServerStopped as it sets dependsOn, or the chain's own stop is what the sync ends up waiting on forever). Gates are therefore set as a set, not patched.
The zero EnqueueOpts releases the task. That is also how a caller un-strands a task it gated onto a pre-assigned _id whose insert then failed.
The bool reports whether the task was still PENDING and so actually re-gated. It is load-bearing, not diagnostic: false means the dispatcher has already claimed the task and it is RUNNING RIGHT NOW. A caller re-gating a pre-existing pending task to keep it behind a task it has not inserted yet MUST treat false as "I lost the race" and abandon the insert — proceeding would leave an ungated syncmods next to a booting game server. Only a caller re-gating a task it just enqueued itself may ignore it.
func ShutdownAgentTaskService ¶
func ShutdownAgentTaskService() error
func StartDispatcher ¶
func StartDispatcher()
func StopDispatcher ¶
func StopDispatcher()
Types ¶
type Assignment ¶
type Assignment struct {
TaskID string
Action string
Data string
LeaseToken string
Attempt int32
MaxAttempts int32
LeaseSeconds int32
}
Assignment is one task pushed down an agent's stream.
type EnqueueOpts ¶
EnqueueOpts carries the optional gates through to the stored task. Zero value means "claimable as soon as the agent is idle", which is what every existing caller wants.
ID lets the caller pre-assign the task's _id. That is what allows a caller to gate an ALREADY-PENDING task behind a task it has not inserted yet: gating first and inserting second means the pending task is never simultaneously ungated and older than the new one, which is the only ordering under which the dispatcher's FIFO claim cannot run them backwards. If the insert then fails, the caller must SetGate the stranded task back to claimable.
type Registry ¶
type Registry struct {
// contains filtered or unexported fields
}
Registry tracks which agents have a live task stream on *this* replica.
func GetRegistry ¶
func GetRegistry() *Registry
func (*Registry) Add ¶
func (r *Registry) Add(agentID bson.ObjectID) (<-chan Assignment, func())
Add registers a stream and returns the receive channel plus a deregister func.
The deregister func removes the entry only if it is still *this* entry, using pointer identity rather than any id the agent supplies. An agent that reconnects before its old server-side stream has torn down would otherwise have the old teardown evict and close the new, healthy stream.
Source Files