queue

package
v0.1.7 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Aug 29, 2026 License: MIT Imports: 19 Imported by: 0

Documentation

Overview

Package queue is the job system: what work exists, how it is enqueued, and how a worker process consumes it.

It replaces BullMQ with River, which moves the queue from Redis into Postgres. That is the substantive change, not an implementation detail. The BullMQ arrangement lost jobs whenever Redis was flushed or evicted under memory pressure, had no transactional relationship with the rows the jobs were about, and needed a separate deduplication mechanism bolted on top. A Postgres-backed queue makes "insert the killmail and enqueue its follow-up work" a single atomic act, and makes the backlog a table anybody can query.

The declarations in internal/jobs stay authoritative for concurrency, retry and backoff. This package turns them into a running client.

Index

Constants

View Source 
const CronConcurrency = 4

CronConcurrency is how many scheduled jobs may run at once.

Low on purpose. Crons are mostly serial maintenance — importers, rebuilds, reconciliations — and several of them touch the same tables. Running a handful concurrently is useful so a slow hourly job does not delay a job that runs every thirty seconds; running many is a way to have two rebuilds of the same aggregate racing each other.

View Source 
const CronQueue = "cron"

CronQueue is where scheduled jobs run.

Separate from the work queues so a slow hourly job cannot occupy the workers that keep the killmail feed moving, and so its concurrency can stay low without throttling anything else.

View Source 
const DefaultBackoff = time.Second

DefaultBackoff applies to a kind with no registry entry — the cron queue, and anything added to River without being declared.

View Source 
const MaxRetryDelay = time.Hour

MaxRetryDelay caps the backoff.

Without a cap the formula runs away: the ESI queues retry ten times from a 30-second base, and 30s × 2⁹ is over four hours. A job that has been waiting four hours to retry is not retrying, it is abandoned with extra steps, and it holds a slot in the backlog the whole time. An hour is long enough to ride out any ESI outage worth riding out.

View Source 
const RetryJitter = 0.15

RetryJitter is how much the delay is randomised, as a fraction.

This is a deliberate departure from BullMQ, whose exponential backoff has no jitter. It matters here because failures in this system are correlated rather than independent: when Tranquility goes down, every in-flight ESI job fails within the same second and, without jitter, every one of them retries in the same second too — repeatedly, in lockstep, for as long as the outage lasts. Spreading them out costs nothing and removes the synchronised stampede.

View Source 
const TQOffline = "offline"

TQOffline is the value that means stop.

View Source 
const TQPollInterval = 10 * time.Second

TQPollInterval is how often the flag is checked.

Ten seconds is a compromise the other way round from most polling: the cost of being late to pause is real (wasted error budget) but the cost of being late to resume is only latency, and the flag itself is only refreshed every thirty seconds by the cron that writes it.

View Source 
const TQStatusKey = "esi:tq:status"

TQStatusKey holds "online" or "offline".

Variables

This section is empty.

Functions

func DispatchCascade 

func DispatchCascade(ctx context.Context, c *Client, cascade Cascade, parent Priority) (int, error)

DispatchCascade enqueues the follow-up work an entity refresh discovered.

The cascade is already deduplicated and filtered by the entities package, so everything here is worth fetching. It inherits the parent's tier rather than resetting to the top, or a single dormant backfill job would spawn immediate work and the priority tiers would stop meaning anything.

func DispatchMany 

func DispatchMany(ctx context.Context, c *Client, batch []river.JobArgs, priority Priority) (int, error)

DispatchMany enqueues a batch in one round trip.

Worth using wherever the count is unbounded — the killmail repair cron finds thousands of missing kills at a time, and inserting those one at a time is thousands of round trips for work that fits in a single statement.

func InsertOptsFor 

func InsertOptsFor(queueName string, priority Priority) *river.InsertOpts

InsertOptsFor builds the insert options for a queue at a priority.

Centralised so that MaxAttempts is never taken from River's default when the registry has an answer, and so uniqueness is on by default. Deduplication being opt-out rather than opt-in is the safer direction: a duplicate that slips through processes a killmail twice and double-counts every statistic it touches, while an unwanted collapse merely defers work by one run.

func Migrate 

func Migrate(ctx context.Context, pool *pgxpool.Pool) ([]string, error)

Migrate applies every pending River migration.

func TQIsOffline 

func TQIsOffline(ctx context.Context, rdb *redis.Client) (bool, error)

TQIsOffline reads the status flag.

An unset key means online. The flag is only ever written to say "offline", so its absence — on a fresh Redis, before the first tq_status run — must not stop every ESI queue in the cluster.

Types

type AchievementAttacker 

type AchievementAttacker struct {
	CharacterID int32 `json:"character_id"`
	ShipGroupID int32 `json:"ship_group_id"`
	FinalBlow   bool  `json:"final_blow"`
}

AchievementAttacker is one participant, reduced to what badges care about.

type AchievementsArgs 

type AchievementsArgs struct {
	KillmailID int64 `json:"killmail_id" river:"unique"`

	TotalValue        float64 `json:"total_value"`
	SystemSecurity    float64 `json:"system_security"`
	HasSecurity       bool    `json:"has_security"`
	IsNPC             bool    `json:"is_npc"`
	IsSolo            bool    `json:"is_solo"`
	VictimShipGroupID int32   `json:"victim_ship_group_id"`
	VictimCharacterID int32   `json:"victim_character_id"`

	Attackers []AchievementAttacker `json:"attackers"`
}

AchievementsArgs awards the badges one killmail earned.

Carries the killmail's shape rather than its id: the fields are a small fixed set the caller already has, so re-reading the mail would be two queries to rebuild what was just in hand.

func (AchievementsArgs) Kind 

func (AchievementsArgs) Kind() string

type AllianceArgs 

type AllianceArgs struct {
	AllianceID int32 `json:"alliance_id" river:"unique"`
}

AllianceArgs fetches and stores one alliance.

func (AllianceArgs) Kind 

func (AllianceArgs) Kind() string

type AnnouncementEventArgs 

type AnnouncementEventArgs struct {
	Payload json.RawMessage `json:"payload"`
}

AnnouncementEventArgs forwards an announcement lifecycle event.

The payload is opaque: it is produced by the frontend and consumed by the relay's clients, and nothing in between has any business reshaping it. Uniqueness keys on the whole payload rather than an id, because the same announcement legitimately produces several events — new, then updated, then expired — and collapsing those would drop all but the first.

func (AnnouncementEventArgs) Kind 

func (AnnouncementEventArgs) Kind() string

type BattleDetectionArgs 

type BattleDetectionArgs struct {
	From time.Time `json:"from" river:"unique"`
	To   time.Time `json:"to" river:"unique"`
}

BattleDetectionArgs finds the battles in one time window.

Both bounds are in the uniqueness key: two windows are the same job only when they cover the same span, and the hourly and daily scans deliberately overlap.

func (BattleDetectionArgs) Kind 

func (BattleDetectionArgs) Kind() string

type CampaignProcessingArgs 

type CampaignProcessingArgs struct {
	CampaignID string `json:"campaign_id" river:"unique"`
}

CampaignProcessingArgs recomputes one campaign's statistics.

func (CampaignProcessingArgs) Kind 

func (CampaignProcessingArgs) Kind() string

type Cascade 

type Cascade struct {
	Characters           []int32
	Corporations         []int32
	Alliances            []int32
	CharacterHistories   []int32
	CorporationHistories []int32
}

Cascade is the follow-up work an entity refresh implies.

Structurally identical to entities.Cascade, and deliberately not that type: having the queue import the entities package and the entities package import the queue would be a cycle, and the alternative — entities knowing how to enqueue — is exactly the coupling that package's doc comment refuses.

type CharacterArgs 

type CharacterArgs struct {
	CharacterID int32 `json:"character_id" river:"unique"`
}

CharacterArgs fetches and stores one character.

func (CharacterArgs) Kind 

func (CharacterArgs) Kind() string

type CharacterHistoryArgs 

type CharacterHistoryArgs struct {
	CharacterID int32 `json:"character_id" river:"unique"`

	// Force overrides the sync marker that otherwise suppresses a fetch whose
	// answer is already known. Deliberately outside the uniqueness key: a
	// forced refresh should collapse against a pending ordinary one rather than
	// queue alongside it.
	Force bool `json:"force,omitempty"`
}

CharacterHistoryArgs fetches one character's corporation history.

func (CharacterHistoryArgs) Kind 

func (CharacterHistoryArgs) Kind() string

type CharacterKillmailArgs 

type CharacterKillmailArgs struct {
	CharacterID int32 `json:"character_id" river:"unique"`
}

CharacterKillmailArgs reads a character's killmails through their own token.

Worth having even though the public feed carries almost everything: a character's own token returns kills that never reached zKillboard, which is the only way to see losses nobody else reported.

func (CharacterKillmailArgs) Kind 

func (CharacterKillmailArgs) Kind() string

type Client 

type Client struct {
	*river.Client[pgx.Tx]
	// contains filtered or unexported fields
}

Client wraps a River client with the registry's settings applied.

func New 

func New(opts Options) (*Client, error)

New builds a client.

Every queue's concurrency, retry count and backoff comes from internal/jobs.Queues rather than being repeated here, so the declarations stay the single source for queue behavior.

func (*Client) Consuming 

func (c *Client) Consuming() []string

Consuming reports the queues this client works, in registry order.

type CommentEventArgs 

type CommentEventArgs struct {
	Payload json.RawMessage `json:"payload"`
}

CommentEventArgs forwards a comment lifecycle event.

func (CommentEventArgs) Kind 

func (CommentEventArgs) Kind() string

type CorporationArgs 

type CorporationArgs struct {
	CorporationID int32 `json:"corporation_id" river:"unique"`
}

CorporationArgs fetches and stores one corporation.

func (CorporationArgs) Kind 

func (CorporationArgs) Kind() string

type CorporationHistoryArgs 

type CorporationHistoryArgs struct {
	CorporationID int32 `json:"corporation_id" river:"unique"`
	Force         bool  `json:"force,omitempty"`
}

CorporationHistoryArgs fetches one corporation's alliance history.

func (CorporationHistoryArgs) Kind 

func (CorporationHistoryArgs) Kind() string

type CorporationKillmailArgs 

type CorporationKillmailArgs struct {
	CorporationID int32 `json:"corporation_id" river:"unique"`
	CharacterID   int32 `json:"character_id"`
}

CorporationKillmailArgs reads a corporation's killmails.

Keyed on the corporation rather than the character: several members may have granted the scope, and their tokens all return the same list. CharacterID is outside the uniqueness key so whichever member's token triggered the job collapses onto one job for the corporation.

func (CorporationKillmailArgs) Kind 

func (CorporationKillmailArgs) Kind() string

type CorporationWalletArgs 

type CorporationWalletArgs struct {
	CorporationID int32 `json:"corporation_id" river:"unique"`

	// Force bypasses the staleness gate. Outside the uniqueness key so a forced
	// sync collapses against a pending scheduled one rather than queueing
	// alongside it and doing the same seven paginated walks twice.
	Force bool `json:"force,omitempty"`
}

CorporationWalletArgs syncs one corporation's wallet.

func (CorporationWalletArgs) Kind 

func (CorporationWalletArgs) Kind() string

type CronArgs 

type CronArgs struct {
	Name string `json:"name" river:"unique"`
}

CronArgs runs one scheduled job.

Every cron shares a single job kind rather than having one kind each. River routes on the kind returned by a zero-valued args struct, so a per-cron kind would need 32 near-identical types and 32 worker registrations to express what is really one dispatch table. The cron's name is in the args instead, where `river_job.args->>'name'` still makes the backlog readable per job.

func (CronArgs) InsertOpts 

func (CronArgs) InsertOpts() river.InsertOpts

InsertOpts sets the cron queue, its retry budget, and its overlap guard.

A cron that fails is retried a couple of times and then left alone until its next tick, which is the natural repair: a job that runs hourly does not need ten retries, because the eleventh attempt would land after the next run has already started.

The uniqueness is what stops a slow cron from piling up on itself. Several run far more often than they are guaranteed to finish — status_update every second, esi_token_sync every thirty — and without this a run that overruns its interval gets a second copy started underneath it, then a third. The TypeScript runner guarded this in memory ("previous run still in progress"); doing it through the queue makes the guarantee cluster-wide instead of per process, which matters now that any number of schedulers can be running.

func (CronArgs) Kind 

func (CronArgs) Kind() string

type Depth 

type Depth struct {
	Queue string `json:"queue"`

	// Available is ready to run now.
	Available int64 `json:"available"`
	// Running is leased by a worker.
	Running int64 `json:"running"`
	// Scheduled is waiting for a future time — a delay or a retry backoff.
	Scheduled int64 `json:"scheduled"`
	// Retryable failed and will be tried again.
	Retryable int64 `json:"retryable"`
	// Discarded exhausted its attempts. These need a human.
	Discarded int64 `json:"discarded"`
	// Cancelled was cancelled deliberately, usually because the work was
	// impossible rather than because it failed.
	Cancelled int64 `json:"cancelled"`
	// Completed succeeded and has not yet been cleaned up.
	Completed int64 `json:"completed"`
}

Depth is one queue's job counts by state.

func Depths 

func Depths(ctx context.Context, pool *pgxpool.Pool) ([]Depth, error)

Depths reads live counts for every queue that has jobs, plus every declared queue, so a queue with nothing in it still appears rather than vanishing.

func StatusDepths 

func StatusDepths(ctx context.Context, pool *pgxpool.Pool) ([]Depth, error)

StatusDepths reads only states that contribute to the published backlog.

Completed and cancelled jobs are retained as diagnostics, but neither is in the BullMQ-shaped status contract. Excluding them is significant at live throughput because completed jobs remain in River for a day.

func (Depth) Pending 

func (d Depth) Pending() int64

Pending is the work still owing. Completed, discarded and cancelled are history and deliberately excluded — a queue with a million completed jobs and nothing waiting is idle, not busy.

func (Depth) Total 

func (d Depth) Total() int64

Total is every job in any state.

type FitExtractArgs 

type FitExtractArgs struct {
	KillmailID int64 `json:"killmail_id" river:"unique"`
}

FitExtractArgs records the fit a killmail's victim was flying.

Only the id: the extractor re-reads the killmail's items, which keeps the payload small and makes a re-run produce exactly what the first run did.

func (FitExtractArgs) Kind 

func (FitExtractArgs) Kind() string

type GraphIngestArgs 

type GraphIngestArgs struct {
	KillmailID int64 `json:"killmail_id" river:"unique"`
}

GraphIngestArgs records one killmail's relationships in the graph.

func (GraphIngestArgs) Kind 

func (GraphIngestArgs) Kind() string

type ImageRefreshArgs 

type ImageRefreshArgs struct {
	EntityKind string `json:"entity_kind" river:"unique"`
	EntityID   int64  `json:"entity_id" river:"unique"`
}

ImageRefreshArgs refreshes one cached character, corporation, or alliance image from CCP. Kind and ID form the uniqueness key so a popular stale image produces one refresh regardless of how many HTTP replicas noticed it.

func (ImageRefreshArgs) Kind 

func (ImageRefreshArgs) Kind() string

type InsertResult 

type InsertResult struct {
	Job *rivertype.JobRow

	// Duplicate is true when an identical job was already queued and this
	// insert collapsed into it. Not an error — it is the deduplication working
	// — but callers counting dispatches need to tell the two apart.
	Duplicate bool
}

InsertResult reports what an insert did.

func Dispatch 

func Dispatch(ctx context.Context, c *Client, args river.JobArgs, priority Priority) (InsertResult, error)

Dispatch enqueues one job.

func DispatchAt 

func DispatchAt(ctx context.Context, c *Client, args river.JobArgs, priority Priority, after time.Duration) (InsertResult, error)

DispatchAt enqueues one job to run no earlier than the given delay.

Used where the TypeScript passed a `delay`: history fetches are spread over a second or so after the character that triggered them, so a burst of killmails does not put a thousand history requests into ESI at the same instant.

type KillmailArgs 

type KillmailArgs struct {
	KillmailID   int64  `json:"killmail_id" river:"unique"`
	KillmailHash string `json:"killmail_hash"`

	// WarID is supplied out of band by the war queue; the public ESI killmail
	// endpoint does not return it.
	WarID int32 `json:"war_id,omitempty"`

	// Killmail is the ESI document, present when the source already had it.
	// R2Z2 embeds it in every feed entry, which is what makes that ingest path
	// cost nothing from the ESI budget — carrying it here is the difference
	// between a fetch per killmail and no fetch at all.
	Killmail *killmail.ESIKillmail `json:"killmail,omitempty"`
}

KillmailArgs processes one killmail.

func (KillmailArgs) Kind 

func (KillmailArgs) Kind() string

type MigrationStatus 

type MigrationStatus struct {
	Applied   []int `json:"applied"`
	Available []int `json:"available"`
	Pending   []int `json:"pending"`
}

MigrationStatus reports where the River schema stands.

func MigrationState 

func MigrationState(ctx context.Context, pool *pgxpool.Pool) (MigrationStatus, error)

MigrationState reads the applied and available versions.

func (MigrationStatus) UpToDate 

func (s MigrationStatus) UpToDate() bool

UpToDate reports whether there is nothing left to apply.

type Options 

type Options struct {
	Pool *pgxpool.Pool

	// Workers is the set of registered job handlers. Nil makes an
	// insert-only client — which is what every process that enqueues work but
	// does not consume it should use, because a client with no workers cannot
	// accidentally start processing a queue another pod owns.
	Workers *river.Workers

	// Queues limits which queues this process consumes. Empty means every
	// declared queue that is not marked as consumed elsewhere.
	Queues []string

	// PeriodicJobs are the scheduled jobs this client inserts if elected
	// leader. Every started worker client must supply the same set because
	// River elects one leader across the whole schema, not only cron workers.
	PeriodicJobs []*river.PeriodicJob

	Logger *slog.Logger
}

Options configure a client.

type Priority 

type Priority int

Priority is a semantic tier, not a number to be invented at each call site.

The four tiers exist so that bulk repair work can never crowd out an explicit request or live entity maintenance. A backfill walking a million characters and the fetch for a character who just appeared on a killmail are both "esi_character" jobs; without tiers the backfill wins by sheer volume and the live killboard shows "Unknown" for minutes. 

const (
	// Immediate is work someone is waiting on — an explicit request from the
	// site or an operator.
	Immediate Priority = iota + 1

	// Live is ordinary maintenance driven by things happening in game now.
	Live

	// RecentBackfill is repair work over the recent past.
	RecentBackfill

	// DormantBackfill is bulk work over everything else, and must never
	// interfere with any of the above.
	DormantBackfill
)

func CascadePriority 

func CascadePriority(parent Priority) Priority

CascadePriority is the tier for follow-up work discovered while running a job, mirroring historyPriorityFromParent().

Follow-up work inherits its parent's urgency rather than resetting to the top. A character discovered during a dormant backfill leads to a corporation fetch that is also dormant backfill; without this, one job at the bottom tier spawns children at the top and the tiers stop meaning anything.

func FromBullMQ 

func FromBullMQ(priority int) Priority

FromBullMQ converts a BullMQ priority to a tier.

Values between the defined tiers round towards the more urgent one, matching historyPriorityFromParent(): a job asking for 3 is treated as recent backfill, not dormant. An absent priority is Live rather than Immediate, because BullMQ's implicit 0 was never a considered choice — it is what a caller that said nothing got, and treating silence as maximum urgency is how the immediate lane fills with routine work.

func (Priority) String 

func (p Priority) String() string

String names the tier for display.

func (Priority) Valid 

func (p Priority) Valid() bool

Valid reports whether a tier is one River will accept.

type RegistryRetryPolicy 

type RegistryRetryPolicy struct {
	// Rand is the jitter source. Nil uses the global one; tests supply their
	// own so the delay is exact and assertions can be about the formula rather
	// than about a range.
	Rand func() float64

	// Now is the clock, injectable for tests.
	Now func() time.Time
}

RegistryRetryPolicy computes retry timing from the queue declarations, so the backoff each queue gets is the one written down in internal/jobs rather than a single client-wide default.

River's retry policy is per client, not per queue, so the per-queue base delay is recovered by looking the job's kind up in the registry.

func (*RegistryRetryPolicy) NextRetry 

func (p *RegistryRetryPolicy) NextRetry(job *rivertype.JobRow) time.Time

NextRetry returns when the job should next be attempted.

The formula is BullMQ's — base × 2^(attempt-1) — so a queue's declared backoff means the same thing after the port as before it.

The delay is measured from now rather than from when the attempt started, which is both what BullMQ does and the only version that works. Measuring from the attempt means a job that runs for longer than its own backoff gets a retry time in the past: the killmails queue has a one-second base delay and a killmail that takes a second to fail lands exactly there. River rejects a past retry time and silently substitutes its own default policy, so the declared per-queue backoff would quietly stop being used at all — which is exactly what happened the first time this ran against the live feed.

type StatsWriterArgs 

type StatsWriterArgs struct {
	KillmailID int64 `json:"killmail_id" river:"unique"`
}

StatsWriterArgs folds one killmail into the aggregate counters.

Only the id: the worker reads the killmail back from the database, so the payload stays tiny and a replay produces exactly the counters the original run did rather than whatever was in flight at dispatch time.

func (StatsWriterArgs) Kind 

func (StatsWriterArgs) Kind() string

type TQGate 

type TQGate struct {
	Client *Client
	Redis  *redis.Client

	// OnChange is called when the gate flips, for logging. Optional.
	OnChange func(offline bool)
	// contains filtered or unexported fields
}

TQGate pauses and resumes the ESI-dependent queues.

func (*TQGate) Watch 

func (g *TQGate) Watch(ctx context.Context) error

Watch blocks, polling the status flag until the context is cancelled.

Redis being unreachable is deliberately not treated as "TQ is offline". A Redis outage would otherwise halt all killmail processing, which is both wrong and the opposite of what an operator wants during an incident: the safer failure is to keep working and let the ESI client's own error-limit handling apply the brakes if it turns out TQ really is down.

type TokenRefreshArgs 

type TokenRefreshArgs struct {
	CharacterID int32 `json:"character_id" river:"unique"`
}

TokenRefreshArgs refreshes one character's SSO token.

func (TokenRefreshArgs) Kind 

func (TokenRefreshArgs) Kind() string

type WarArgs 

type WarArgs struct {
	WarID int32 `json:"war_id" river:"unique"`

	// MetadataOnly skips walking the war's killmail list. Set by the repair
	// sweep, which fills in wars referenced by killmails already stored — so
	// paging ESI for those killmails would spend the request budget to discover
	// nothing, across hundreds of thousands of finished wars.
	MetadataOnly bool `json:"metadata_only,omitempty"`
}

WarArgs fetches one war.

func (WarArgs) Kind 

func (WarArgs) Kind() string

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.