aquifer

package module

Go to main page

Versions in this module

v0

v0.11.0

Sep 23, 2026

Changes in this version

+ const Version

+ var ErrAquiferDraining = errors.New("aquifer is draining")

+ var ErrJobResultNotFound = errors.New("job result not found")

+ var ErrWebSocketClientLimit = errors.New("websocket client connection limit reached")

+ var ErrWebSocketProtocol = errors.New("invalid aqueduct websocket message")

+ var ErrWebSocketReplayGap = errors.New("websocket replay cursor is older than retained history")

+ var ErrWebSocketSchedulerClosed = errors.New("websocket scheduler closed")

+ var ErrWebSocketWaitingLimit = errors.New("websocket waiting connection limit reached")

+ func ShutdownHTTPServer(server *http.Server)

type Aquifer

+ func (a *Aquifer) BeginDrain(webSocketGrace time.Duration) bool

+ func (a *Aquifer) GetJobResult(userID, idempotentKey string) (JobResult, error)

+ func (a *Aquifer) IsDraining() bool

+ func (a *Aquifer) SetWebSocketManager(manager *WebSocketManager)

type ClusterConfig

+ PruneTTL time.Duration

type ClusterRouter

+ func (r *ClusterRouter) PruneMember(id string)

+ func (r *ClusterRouter) RankedOwners(key string) []ClusterMember

type EnqueueResult

+ ResultKey string

+ type JobResult struct

+ Body string

+ BodyTruncated bool

+ ContentType string

+ JobID string

+ RecordedAt int64

+ ResponseStatus int

+ Source string

+ Status Status

+ type JobResultReader interface

+ LookupResult func(hash string) (JobResult, bool)

+ type JobResultRecorder interface

+ RecordResult func(hash string, result JobResult) (string, bool)

+ type LifecycleState string

+ const LifecycleStateActive

+ const LifecycleStateDraining

+ const LifecycleStateOffline

+ type RedisWebSocketStreamStore struct

+ func NewRedisWebSocketStreamStore(rawURL, prefix string, maxEvents int64, ttl time.Duration) (*RedisWebSocketStreamStore, error)

+ func (s *RedisWebSocketStreamStore) Append(ctx context.Context, sessionID, direction string, message WebSocketEnvelope) (string, error)

+ func (s *RedisWebSocketStreamStore) CheckCursor(ctx context.Context, sessionID, after string) error

+ func (s *RedisWebSocketStreamStore) Close() error

+ func (s *RedisWebSocketStreamStore) Ping(ctx context.Context) error

+ func (s *RedisWebSocketStreamStore) ReadAfter(ctx context.Context, sessionID, after string, count int64, block time.Duration) ([]WebSocketStreamEvent, error)

type Registry

+ func (r *Registry) BeginDrain()

+ func (r *Registry) FlushForShutdown(ctx context.Context) bool

+ func (r *Registry) ReportOffline(ctx context.Context)

+ func (r *Registry) WaitIdle(ctx context.Context) error

type RemoteIdempotencyConfig

+ ResultEnabled bool

+ ResultMaxBytes int64

+ ResultPrefix string

type RemoteIdempotencyEntry

+ ResultKey string

type Runtime

+ Shutdown ShutdownConfig

+ WebSockets *WebSocketManager

+ func (r *Runtime) BeginDrain()

+ func (r *Runtime) Drain(ctx context.Context) error

type RuntimeOptions

+ ShutdownConfig *ShutdownConfig

+ WebSocketConfig *WebSocketConfig

+ WebSocketStore WebSocketStreamStore

+ type ShutdownConfig struct

+ Quiesce time.Duration

+ Timeout time.Duration

+ WebSocketGrace time.Duration

+ func LoadShutdownConfig() ShutdownConfig

type ValkeyRemoteIdempotency

+ func (v *ValkeyRemoteIdempotency) LookupResult(hash string) (JobResult, bool)

+ func (v *ValkeyRemoteIdempotency) RecordResult(hash string, result JobResult) (string, bool)

+ type WebSocketAdmission struct

+ Position int

+ func (a *WebSocketAdmission) Cancel()

+ func (a *WebSocketAdmission) Wait(ctx context.Context, onPosition func(int) error) (func(), error)

+ type WebSocketConfig struct

+ Enabled bool

+ HandshakeTimeout time.Duration

+ MaxMessageBytes int64

+ ReadBatch int64

+ ReadBlock time.Duration

+ ReconnectMax time.Duration

+ RedisURL string

+ Scheduler WebSocketSchedulerConfig

+ StreamMaxEvents int64

+ StreamPrefix string

+ StreamTTL time.Duration

+ func LoadWebSocketConfig() WebSocketConfig

+ type WebSocketEnvelope struct

+ CausedBy string

+ ConnectRPS *float64

+ Generation int64

+ MaxConnections *int

+ MessageID string

+ Payload json.RawMessage

+ Position int

+ Reason string

+ RetryAfterMS int64

+ State string

+ StreamID string

+ Type string

+ type WebSocketManager struct

+ func NewWebSocketManager(cfg WebSocketConfig, store WebSocketStreamStore) *WebSocketManager

+ func NewWebSocketManagerFromConfig(cfg WebSocketConfig) (*WebSocketManager, error)

+ func (m *WebSocketManager) BeginDrain(grace time.Duration)

+ func (m *WebSocketManager) Close()

+ func (m *WebSocketManager) IsDraining() bool

+ func (m *WebSocketManager) ServeHTTP(w http.ResponseWriter, r *http.Request)

+ func (m *WebSocketManager) Snapshot() map[string]any

+ func (m *WebSocketManager) WaitDrained(ctx context.Context) error

+ type WebSocketScheduler struct

+ func NewWebSocketScheduler(cfg WebSocketSchedulerConfig) *WebSocketScheduler

+ func (s *WebSocketScheduler) AcquireClient() error

+ func (s *WebSocketScheduler) Close()

+ func (s *WebSocketScheduler) EnqueueUpstream() (*WebSocketAdmission, error)

+ func (s *WebSocketScheduler) ReleaseClient()

+ func (s *WebSocketScheduler) ReportFailure()

+ func (s *WebSocketScheduler) ReportSuccess()

+ func (s *WebSocketScheduler) Snapshot() WebSocketSchedulerSnapshot

+ func (s *WebSocketScheduler) UpdateCapacity(maxConnections *int, connectRPS *float64)

+ type WebSocketSchedulerConfig struct

+ ConnectRPS float64

+ MaxClients int

+ MaxUpstreams int

+ MaxWaiting int

+ SlowStartRPS float64

+ type WebSocketSchedulerSnapshot struct

+ ActiveUpstreams int

+ Clients int

+ ConnectRPS float64

+ CurrentRampRPS float64

+ EffectiveConnectRPS float64

+ EffectiveMax int

+ MaxClients int

+ MaxUpstreams int

+ SlowStartRPS float64

+ Waiting int

+ type WebSocketStreamEvent struct

+ Direction string

+ Envelope WebSocketEnvelope

+ RecordedAt int64

+ StreamID string

+ type WebSocketStreamStore interface

+ Append func(ctx context.Context, sessionID, direction string, message WebSocketEnvelope) (string, error)

+ CheckCursor func(ctx context.Context, sessionID, after string) error

+ Close func() error

+ Ping func(ctx context.Context) error

+ ReadAfter func(ctx context.Context, sessionID, after string, count int64, block time.Duration) ([]WebSocketStreamEvent, error)

v0.10.0

Sep 7, 2026

Changes in this version

type AccountQueue

+ func (q *AccountQueue) Stop()

type Aquifer

+ func (a *Aquifer) Close()

+ func (a *Aquifer) SetClusterRouter(router *ClusterRouter)

+ func (a *Aquifer) SetRemoteIdempotency(remote RemoteIdempotency)

+ type ClusterConfig struct

+ Enabled bool

+ Load float64

+ Members []ClusterMember

+ PartitionCount int

+ ReplicationFactor int

+ Self ClusterMember

+ func LoadClusterConfig() ClusterConfig

+ type ClusterMember struct

+ Address string

+ ID string

+ func (m ClusterMember) String() string

+ type ClusterRouter struct

+ func NewClusterRouter(cfg ClusterConfig) *ClusterRouter

+ func (r *ClusterRouter) IsOwner(key string) bool

+ func (r *ClusterRouter) OwnerFor(key string) (ClusterMember, bool)

+ func (r *ClusterRouter) Snapshot() map[string]any

type DrainConfig

+ BatchEnabled bool

+ BatchIntervalSeconds int64

+ BatchMaxEvents int

+ Sink string

+ type DrainEvent struct

+ HashKey string

+ JobID string

+ RecordedAt int64

+ Sequence int64

+ Status Status

type FlyRegionAdapter

+ func (a *FlyRegionAdapter) Close()

type JobStore

+ AcknowledgeDrainEventsThrough func(sequence int64)

+ ListDrainEvents func(limit int) []DrainEvent

type L8Registry

+ func (r *L8Registry) Close()

type PebbleStore

+ func (s *PebbleStore) AcknowledgeDrainEventsThrough(sequence int64)

+ func (s *PebbleStore) ListDrainEvents(limit int) []DrainEvent

+ type RegistrationConfig struct

+ IntervalSeconds int64

+ Port string

+ URL string

+ func LoadRegistrationConfig() RegistrationConfig

+ func (cfg RegistrationConfig) Enabled() bool

type Registry

+ func (r *Registry) Close()

+ func (r *Registry) SetDrainRemote(remote RemoteIdempotency)

+ type RemoteIdempotency interface

+ Lookup func(hash string) (RemoteIdempotencyEntry, bool)

+ Record func(events []DrainEvent) bool

+ func NewValkeyRemoteIdempotency(cfg RemoteIdempotencyConfig) RemoteIdempotency

+ type RemoteIdempotencyConfig struct

+ Enabled bool

+ Prefix string

+ TTLSeconds int64

+ Timeout time.Duration

+ URL string

+ func LoadRemoteIdempotencyConfig() RemoteIdempotencyConfig

+ type RemoteIdempotencyEntry struct

+ JobID string

+ RecordedAt int64

+ Source string

+ Status Status

type Runtime

+ func (r *Runtime) Close()

type RuntimeOptions

+ ClusterRouter *ClusterRouter

+ RemoteIdempotency RemoteIdempotency

type Store

+ func (s *Store) AcknowledgeDrainEventsThrough(sequence int64)

+ func (s *Store) ListDrainEvents(limit int) []DrainEvent

type URLWorker

+ func (w *URLWorker) Stop()

+ type ValkeyRemoteIdempotency struct

+ func (v *ValkeyRemoteIdempotency) Lookup(hash string) (RemoteIdempotencyEntry, bool)

+ func (v *ValkeyRemoteIdempotency) Record(events []DrainEvent) bool

v0.9.0

Sep 2, 2026

Changes in this version

type AccountQueue

+ func (q *AccountQueue) Active() bool

type Aquifer

+ func (a *Aquifer) AttemptDirect(ctx context.Context, req JobRequest, timeout time.Duration) ProxyOutcome

+ func (a *Aquifer) Dispatch(job *Job, accountQueueHeader string)

+ func (a *Aquifer) PrepareJob(req JobRequest) (job *Job, duplicate *EnqueueResult, err error)

+ func (a *Aquifer) SetRegionAdapter(adapter RegionAdapter)

+ type FlyRegionAdapter struct

+ func NewFlyRegionAdapter() *FlyRegionAdapter

+ func (a *FlyRegionAdapter) LiveRegions() []string

+ func (a *FlyRegionAdapter) SelfRegion() string

type Job

+ OriginMachineID string

+ OriginRegion string

+ RerouteCount int

+ VisitedRegions []string

type JobRequest

+ DirectOnly bool

+ OriginMachineID string

+ OriginRegion string

+ RerouteCount int

+ VisitedRegions []string

+ type NoopRegionAdapter struct

+ func (NoopRegionAdapter) LiveRegions() []string

+ func (NoopRegionAdapter) SelfRegion() string

+ type ProxyFallbackInfo struct

+ Reason string

+ Status int

+ type ProxyOutcome struct

+ Body []byte

+ Direct bool

+ Duplicate bool

+ Err error

+ ExistingJob *Job

+ FallbackReason string

+ FallbackStatus int

+ Header http.Header

+ Job *Job

+ RelayFrom *http.Response

+ RerouteRegion string

+ Status int

+ type RedirectExhaustedError struct

+ JobID string

+ func (e *RedirectExhaustedError) Error() string

+ type RegionAdapter interface

+ LiveRegions func() []string

+ SelfRegion func() string

type RuntimeOptions

+ RegionAdapter RegionAdapter

type URLWorker

+ func (w *URLWorker) BreakerKind() string

+ func (w *URLWorker) BreakerOpen() bool

+ func (w *URLWorker) QueueActive() bool

+ func (w *URLWorker) TripBreaker(cooldown time.Duration, kind string)

v0.8.0

Aug 25, 2026

v0.7.0

Aug 24, 2026

Changes in this version

type Registry

+ func (r *Registry) EnqueueWebhook(originalJobID, userID, webhookURL string, payload map[string]any)

v0.6.0

Aug 24, 2026

Changes in this version

+ type DrainConfig struct

+ Enabled bool

+ TimerSeconds int64

+ WebhookURL string

+ func LoadDrainConfig() DrainConfig

+ type DrainState string

+ const DrainStateActive

+ const DrainStateDraining

+ const DrainStateUnassigned

type JobStore

+ ClearIdempotentKeys func()

+ ListIdempotentKeys func() []LedgerEntry

+ type LedgerEntry struct

+ HashKey string

+ JobID string

+ Status Status

type MetricsAdapter

+ DrainFlushFailed func(instanceKey string, ledgerSize int)

+ DrainFlushSucceeded func(instanceKey string, ledgerSize int)

type NoopMetricsAdapter

+ func (NoopMetricsAdapter) DrainFlushFailed(instanceKey string, ledgerSize int)

+ func (NoopMetricsAdapter) DrainFlushSucceeded(instanceKey string, ledgerSize int)

type PebbleStore

+ func (s *PebbleStore) ClearIdempotentKeys()

+ func (s *PebbleStore) ListIdempotentKeys() []LedgerEntry

type Registry

+ func (r *Registry) ConfigureDrain(cfg DrainConfig)

+ func (r *Registry) DrainSnapshot() map[string]any

+ func (r *Registry) DrainState() DrainState

type RuntimeOptions

+ DrainConfig *DrainConfig

type Store

+ func (s *Store) ClearIdempotentKeys()

+ func (s *Store) ListIdempotentKeys() []LedgerEntry

v0.5.0

Aug 23, 2026

Changes in this version

type RuntimeOptions

+ Store JobStore

v0.4.0

Aug 15, 2026

Changes in this version

type AccountQueue

+ func (q *AccountQueue) Throttle(rps float64)

type Aquifer

+ func (a *Aquifer) RegisterPoolMember(poolID, memberID, address string, declaredRPS float64, ...) error

type Job

+ PoolID string

type JobRequest

+ PoolID string

+ type JobStore interface

+ CheckOrInsert func(job *Job) (string, bool)

+ Close func() error

+ Counts func() StoreCounts

+ DeleteJob func(jobID string)

+ GetJob func(jobID string) *Job

+ GetQueuedJobs func() []*Job

+ MarkInFlight func(jobID string)

+ Path func() string

+ RecoverInFlight func(queueKey string) []*Job

+ SetQueueKey func(jobID, queueKey string)

+ UpdateStatus func(jobID string, status Status)

+ func NewJobStore(path string) JobStore

+ type PebbleStore struct

+ func NewPebbleStore(path string) *PebbleStore

+ func (s *PebbleStore) CheckOrInsert(job *Job) (string, bool)

+ func (s *PebbleStore) Close() error

+ func (s *PebbleStore) Counts() StoreCounts

+ func (s *PebbleStore) DeleteJob(jobID string)

+ func (s *PebbleStore) GetJob(jobID string) *Job

+ func (s *PebbleStore) GetQueuedJobs() []*Job

+ func (s *PebbleStore) MarkInFlight(jobID string)

+ func (s *PebbleStore) Path() string

+ func (s *PebbleStore) RecoverInFlight(queueKey string) []*Job

+ func (s *PebbleStore) SetQueueKey(jobID, queueKey string)

+ func (s *PebbleStore) UpdateStatus(jobID string, status Status)

+ type Pool struct

+ func (p *Pool) Pick() *PoolMember

+ func (p *Pool) RecordFailure(id string) bool

+ func (p *Pool) RecordSuccess(id string)

+ func (p *Pool) Register(id, address string, declaredRPS float64, heartbeatInterval time.Duration)

+ func (p *Pool) Size() int

+ func (p *Pool) Snapshot() []map[string]any

+ func (p *Pool) TotalCapacity() float64

+ type PoolMember struct

+ Address string

+ DeclaredRPS float64

+ HeartbeatInterval time.Duration

+ ID string

+ LastHeartbeat time.Time

+ func (m *PoolMember) Reputation() float64

+ type PoolRegistry struct

+ func NewPoolRegistry() *PoolRegistry

+ func (pr *PoolRegistry) Get(poolID string) *Pool

+ func (pr *PoolRegistry) Register(poolID, memberID, address string, declaredRPS float64, ...)

+ func (pr *PoolRegistry) Snapshot() map[string]any

+ func (pr *PoolRegistry) Stop()

type Runtime

+ Pools *PoolRegistry

type Store

+ func (s *Store) Close() error

v0.3.0

Jul 24, 2026

Changes in this version

+ type AdmissionController struct

+ func NewAdmissionController(limits AdmissionLimits, dbPath string) *AdmissionController

+ func (c *AdmissionController) AnyLimitConfigured() bool

+ func (c *AdmissionController) Check() AdmissionDecision

+ func (c *AdmissionController) MaxBodyBytes() int64

+ func (c *AdmissionController) RetryAfterSeconds() int

+ func (c *AdmissionController) Snapshot() map[string]any

+ type AdmissionDecision struct

+ Allowed bool

+ Current int64

+ Limit int64

+ Reason string

+ type AdmissionLimits struct

+ DBMaxBytes int64

+ MaxBodyBytes int64

+ MemoryLimitMB int64

+ RetryAfterSeconds int

+ func LoadAdmissionLimits() AdmissionLimits

+ type AdmissionRejectedError struct

+ Decision AdmissionDecision

+ func (e *AdmissionRejectedError) Error() string

type Aquifer

+ func (a *Aquifer) AdmissionSnapshot() map[string]any

+ func (a *Aquifer) MaxBodyBytes() int64

+ func (a *Aquifer) RetryAfterSeconds() int

type JobRequest

+ AccountQueueMode string

type Runtime

+ Admission *AdmissionController

type RuntimeOptions

+ AdmissionLimits *AdmissionLimits

type Store

+ func (s *Store) DeleteJob(jobID string)

v0.2.0

Jun 6, 2026

Changes in this version

+ var ErrJobNotFound = errors.New("job not found")

+ func RunAdapter(ctx context.Context, adapter FrameworkAdapter, opts RuntimeOptions) error

+ type AccountQueue struct

+ func NewAccountQueue(key, upstream string, rps float64, maxConc int, store *Store, broker *Broker, ...) *AccountQueue

+ func (q *AccountQueue) Enqueue(job *Job)

+ func (q *AccountQueue) RPS() float64

+ type Aquifer struct

+ func NewAquifer(store *Store, registry *Registry, broker *Broker, l8 *L8Registry) *Aquifer

+ func (a *Aquifer) Enqueue(req JobRequest) (EnqueueResult, error)

+ func (a *Aquifer) GetJob(id string) (*Job, error)

+ func (a *Aquifer) HandleL8Challenge(req L8ChallengeReq) (*L8ChallengeResp, error)

+ func (a *Aquifer) Health() map[string]any

+ func (a *Aquifer) L8Metadata(host string) L8Meta

+ func (a *Aquifer) SubscribeJob(id string) (*Job, <-chan SSEEvent, func(), error)

+ type Broker struct

+ func NewBroker() *Broker

+ func (b *Broker) Publish(jobID string, event SSEEvent)

+ func (b *Broker) Subscribe(jobID string) (<-chan SSEEvent, func())

+ type Config struct

+ Defaults RateConfig

+ Upstreams map[string]RateConfig

+ func LoadConfig(path string) *Config

+ func (c *Config) ForURL(rawURL string) RateConfig

+ type EnqueueResult struct

+ Duplicate bool

+ JobID string

+ Status Status

+ type FrameworkAdapter interface

+ Name func() string

+ Start func(ctx context.Context, aquifer *Aquifer) error

+ type HTTPAdapter struct

+ func NewHTTPAdapter(addr string) *HTTPAdapter

+ func (a *HTTPAdapter) Name() string

+ func (a *HTTPAdapter) Start(ctx context.Context, aquifer *Aquifer) error

+ type Job struct

+ Body string

+ CreatedAt int64

+ Headers map[string]string

+ ID string

+ IdempotentKey string

+ Method string

+ Status Status

+ URL string

+ UserID string

+ WebhookURL string

+ func NewJob(r *JobRequest) *Job

+ type JobRequest struct

+ Body string

+ Headers map[string]string

+ IdempotentKey string

+ Method string

+ URL string

+ UserID string

+ WebhookURL string

+ func (r *JobRequest) Validate() string

+ type L8ChallengeReq struct

+ ChallengeID string

+ Nonce string

+ SenderPublicKey string

+ Signature string

+ Timestamp int64

+ type L8ChallengeResp struct

+ ChallengeID string

+ Nonce string

+ ReceiverPublicKey string

+ ReceiverSignature string

+ type L8Meta struct

+ Capabilities []string

+ ChallengeEndpoint string

+ ProtocolVersion string

+ PublicKey string

+ ServiceName string

+ SpecURL string

+ SupportedAlgos []string

+ type L8Registry struct

+ PubB64 string

+ func NewL8Registry(keyPath, trustDir string) *L8Registry

+ func (r *L8Registry) EnsureTrust(webhookURL string)

+ func (r *L8Registry) HandleChallenge(req L8ChallengeReq) (*L8ChallengeResp, error)

+ func (r *L8Registry) IsTrusted(webhookURL string) bool

+ func (r *L8Registry) Meta(host string) L8Meta

+ func (r *L8Registry) SignHeaders(body []byte) map[string]string

+ type MCPStdioAdapter struct

+ func NewMCPStdioAdapter(in io.Reader, out io.Writer) *MCPStdioAdapter

+ func (a *MCPStdioAdapter) Name() string

+ func (a *MCPStdioAdapter) Start(ctx context.Context, aquifer *Aquifer) error

+ type MetricsAdapter interface

+ FlowRate func(upstream string, rps float64)

+ JobCompleted func(userID, upstream string, durationMs int64)

+ JobDispatched func(userID, upstream string)

+ JobFailed func(userID, upstream string, reason string)

+ JobQueued func(userID, upstream string)

+ QueueDepth func(upstream string, depth int)

+ WebhookDelivered func(url string, attempt int)

+ WebhookFailed func(url string, attempts int)

+ type NoopMetricsAdapter struct

+ func (NoopMetricsAdapter) FlowRate(upstream string, rps float64)

+ func (NoopMetricsAdapter) JobCompleted(userID, upstream string, durationMs int64)

+ func (NoopMetricsAdapter) JobDispatched(userID, upstream string)

+ func (NoopMetricsAdapter) JobFailed(userID, upstream string, reason string)

+ func (NoopMetricsAdapter) JobQueued(userID, upstream string)

+ func (NoopMetricsAdapter) QueueDepth(upstream string, depth int)

+ func (NoopMetricsAdapter) WebhookDelivered(url string, attempt int)

+ func (NoopMetricsAdapter) WebhookFailed(url string, attempts int)

+ type RateConfig struct

+ MaxConcurrent int

+ RPS float64

+ type Registry struct

+ func NewRegistry(store *Store, cfg *Config, broker *Broker, l8 *L8Registry, ...) *Registry

+ func (r *Registry) Enqueue(job *Job)

+ func (r *Registry) JobDispatched()

+ func (r *Registry) JobDone()

+ type Runtime struct

+ Aquifer *Aquifer

+ Broker *Broker

+ Config *Config

+ L8 *L8Registry

+ Registry *Registry

+ Store *Store

+ func NewRuntime(opts RuntimeOptions) *Runtime

+ func (r *Runtime) DBPath() string

+ func (r *Runtime) RecoverQueuedJobs(dbPath string)

+ type RuntimeOptions struct

+ Config *Config

+ ConfigPath string

+ DBPath string

+ L8KeyPath string

+ L8TrustDir string

+ Metrics MetricsAdapter

+ type SSEEvent struct

+ Data map[string]any

+ Event string

+ type Server struct

+ func NewServer(aquifer *Aquifer) *Server

+ func (s *Server) Routes() http.Handler

+ type Status string

+ const StatusCompleted

+ const StatusFailed

+ const StatusInFlight

+ const StatusQueued

+ type Store struct

+ func NewStore(path string) *Store

+ func (s *Store) CheckOrInsert(job *Job) (string, bool)

+ func (s *Store) Counts() StoreCounts

+ func (s *Store) GetJob(jobID string) *Job

+ func (s *Store) GetQueuedJobs() []*Job

+ func (s *Store) MarkInFlight(jobID string)

+ func (s *Store) Path() string

+ func (s *Store) RecoverInFlight(queueKey string) []*Job

+ func (s *Store) SetQueueKey(jobID, queueKey string)

+ func (s *Store) UpdateStatus(jobID string, status Status)

+ type StoreCounts struct

+ QueueDepth int64

+ TotalJobs int64

+ type URLWorker struct

+ func NewURLWorker(domain string, rps float64, maxConc int, store *Store, broker *Broker, ...) *URLWorker

+ func (w *URLWorker) Enqueue(job *Job)

v0.1.0

May 20, 2026

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL