engine

package
v0.1.0 Latest Latest
Warning

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

 Go to latest Published: Aug 3, 2026 License: Apache-2.0 Imports: 30 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func CanonicalizeURL 

func CanonicalizeURL(rawURL string) string

CanonicalizeURL normalizes a URL for deduplication: - lowercases scheme and host - removes fragment - sorts query parameters - removes trailing slash (except root) - removes default ports (80 for http, 443 for https)

Types

type AutoscaleConfig 

type AutoscaleConfig struct {
	// MinConcurrency is the minimum number of workers.
	MinConcurrency int

	// MaxConcurrency is the maximum number of workers.
	MaxConcurrency int

	// ScaleUpThreshold: if system load (0.0-1.0) is below this, add workers.
	ScaleUpThreshold float64

	// ScaleDownThreshold: if system load is above this, remove workers.
	ScaleDownThreshold float64

	// CheckInterval is how often to evaluate scaling decisions.
	CheckInterval time.Duration

	// CooldownPeriod is the minimum time between scaling actions.
	CooldownPeriod time.Duration
}

AutoscaleConfig configures the autoscaled worker pool.

func DefaultAutoscaleConfig 

func DefaultAutoscaleConfig() *AutoscaleConfig

DefaultAutoscaleConfig returns sensible defaults.

type AutoscaledPool 

type AutoscaledPool struct {
	// contains filtered or unexported fields
}

AutoscaledPool dynamically adjusts worker concurrency based on system load. Inspired by Crawlee's AutoscaledPool — ramp up when idle, back off when strained.

func NewAutoscaledPool 

func NewAutoscaledPool(cfg *AutoscaleConfig, queueSizeFn func() int, logger *slog.Logger) *AutoscaledPool

NewAutoscaledPool creates an autoscaled pool.

func (*AutoscaledPool) CurrentConcurrency 

func (p *AutoscaledPool) CurrentConcurrency() int

CurrentConcurrency returns the current worker count.

func (*AutoscaledPool) DesiredConcurrency 

func (p *AutoscaledPool) DesiredConcurrency() int

DesiredConcurrency returns the target worker count after the last evaluation.

func (*AutoscaledPool) Evaluate 

func (p *AutoscaledPool) Evaluate() ScaleDecision

Evaluate checks current system load and decides whether to scale up or down. Returns the new desired concurrency. Call this periodically from a monitoring goroutine.

func (*AutoscaledPool) RecordWorkerActive 

func (p *AutoscaledPool) RecordWorkerActive()

RecordWorkerActive increments the active worker counter.

func (*AutoscaledPool) RecordWorkerIdle 

func (p *AutoscaledPool) RecordWorkerIdle()

RecordWorkerIdle increments the idle worker counter.

func (*AutoscaledPool) SetWorkerCounts 

func (p *AutoscaledPool) SetWorkerCounts(active, idle int)

SetWorkerCounts sets the current worker counts directly.

type BloomDeduplicator 

type BloomDeduplicator struct {
	// contains filtered or unexported fields
}

BloomDeduplicator is a memory-bounded, lossy deduplicator.

It holds a Bloom filter and nothing else. That is the point: the earlier version of this type kept a Bloom filter *and* a complete exact set, so it used strictly more memory than the plain map it was meant to improve on — the fast negative lookup was real, the advertised 10-100x memory saving was not.

The trade

Bloom filters have no false negatives and some false positives. For crawl dedup that means: a URL genuinely seen is always reported seen (never re-crawled), and a URL never seen is occasionally reported seen — and therefore silently skipped. At the default 1% rate, roughly one URL in a hundred is dropped.

That is a real cost and it is why this is not the default. It is the right trade only when the crawl is large enough that an exact set will not fit, where the alternative is not "crawl everything" but "run out of memory".

Export and Import are deliberately unsupported: a Bloom filter cannot enumerate its members, so a checkpoint cannot round-trip through one. Resuming a Bloom crawl restarts with an empty filter, which re-crawls rather than drops — the safer direction of the two.

func NewBloomDeduplicator 

func NewBloomDeduplicator(expectedElements int, fpRate float64) *BloomDeduplicator

NewBloomDeduplicator creates a Bloom-backed deduplicator sized for the expected number of URLs, at the given false-positive rate. A rate of zero uses 1%.

func (*BloomDeduplicator) Count 

func (bd *BloomDeduplicator) Count() int

Count returns the number of URLs added.

func (*BloomDeduplicator) Export 

func (bd *BloomDeduplicator) Export() []string

Export returns nil: a Bloom filter cannot enumerate its members, so there is nothing to checkpoint. See the type comment.

func (*BloomDeduplicator) Import 

func (bd *BloomDeduplicator) Import([]string)

Import is a no-op, for the same reason as Export.

func (*BloomDeduplicator) IsSeen 

func (bd *BloomDeduplicator) IsSeen(rawURL string) bool

IsSeen reports whether a URL has been claimed, subject to the filter's false-positive rate.

func (*BloomDeduplicator) MarkIfUnseen 

func (bd *BloomDeduplicator) MarkIfUnseen(rawURL string) bool

MarkIfUnseen atomically claims a URL, reporting whether it was new.

func (*BloomDeduplicator) MemoryStats 

func (bd *BloomDeduplicator) MemoryStats() map[string]any

MemoryStats returns memory usage information.

func (*BloomDeduplicator) Reset 

func (bd *BloomDeduplicator) Reset()

Reset clears the filter.

type BloomFilter 

type BloomFilter struct {
	// contains filtered or unexported fields
}

BloomFilter is a space-efficient probabilistic data structure for URL deduplication. At scale (millions of URLs), a Bloom filter uses ~10 bits per element instead of storing full URL hashes, reducing memory by 10-100x while maintaining <1% false positive rate.

func NewBloomFilter 

func NewBloomFilter(expectedElements int, fpRate float64) *BloomFilter

NewBloomFilter creates a new Bloom filter for the expected number of elements with the desired false positive rate (e.g., 0.01 = 1%).

Example: 1M URLs at 1% FP rate → ~1.2 MB memory (vs ~64 MB for map-based dedup)

func (*BloomFilter) Add 

func (bf *BloomFilter) Add(url string)

Add inserts a URL into the Bloom filter.

func (*BloomFilter) Contains 

func (bf *BloomFilter) Contains(url string) bool

Contains checks if a URL might be in the set. Returns true if the URL is PROBABLY in the set (with false positive rate). Returns false if the URL is DEFINITELY NOT in the set.

func (*BloomFilter) Count 

func (bf *BloomFilter) Count() uint64

Count returns the number of elements added (not unique, since we can't know exactly).

func (*BloomFilter) EstimatedFPRate 

func (bf *BloomFilter) EstimatedFPRate() float64

EstimatedFPRate returns the current estimated false positive rate.

func (*BloomFilter) MemoryUsageBytes 

func (bf *BloomFilter) MemoryUsageBytes() uint64

MemoryUsageBytes returns the approximate memory usage in bytes.

func (*BloomFilter) Reset 

func (bf *BloomFilter) Reset()

Reset clears the Bloom filter.

type CheckpointManager 

type CheckpointManager struct {
	// contains filtered or unexported fields
}

CheckpointManager handles saving and loading crawl state for pause/resume.

func NewCheckpointManager 

func NewCheckpointManager(interval time.Duration) *CheckpointManager

NewCheckpointManager creates a new CheckpointManager.

func (*CheckpointManager) Clean 

func (cm *CheckpointManager) Clean() error

Clean removes the checkpoint file.

func (*CheckpointManager) Dir 

func (cm *CheckpointManager) Dir() string

Dir returns the directory checkpoints are written to.

func (*CheckpointManager) HasCheckpoint 

func (cm *CheckpointManager) HasCheckpoint() bool

HasCheckpoint returns true if a checkpoint file exists.

func (*CheckpointManager) Load 

func (cm *CheckpointManager) Load(frontier *Frontier, dedup Deduper, stats *Stats) error

Load reads a checkpoint from disk and restores crawl state.

func (*CheckpointManager) Save 

func (cm *CheckpointManager) Save(frontier *Frontier, dedup Deduper, stats *Stats) error

Save serializes the current crawl state to disk.

type CircuitBreaker 

type CircuitBreaker struct {
	// contains filtered or unexported fields
}

CircuitBreaker stops a crawl from hammering a domain that is consistently failing.

Without one, a site that goes down absorbs the full request budget: every URL queued for it is attempted, retried to exhaustion, and fails, while the crawler reports the wasted work as ordinary errors. The breaker turns that into a fast rejection and gives the site room to recover.

Sized by the same LRU discipline as the throttler, since a broad crawl sees unboundedly many domains.

func NewCircuitBreaker 

func NewCircuitBreaker(threshold int, cooldown time.Duration, maxSize int) *CircuitBreaker

NewCircuitBreaker builds a breaker. A threshold of zero or less disables it.

func (*CircuitBreaker) Allow 

func (cb *CircuitBreaker) Allow(domain string) bool

Allow reports whether a request to domain may be attempted.

An open circuit lets exactly one probe through once the cooldown has elapsed. Letting several through would re-hammer a site that has not recovered, and the answer from one probe is as informative as the answer from fifty.

func (*CircuitBreaker) Enabled 

func (cb *CircuitBreaker) Enabled() bool

Enabled reports whether the breaker will reject anything.

func (*CircuitBreaker) RecordFailure 

func (cb *CircuitBreaker) RecordFailure(domain string) breakerState

RecordFailure reports a failed fetch. Returns the resulting state, so the caller can log the closed→open transition rather than every failure.

func (*CircuitBreaker) RecordSuccess 

func (cb *CircuitBreaker) RecordSuccess(domain string)

RecordSuccess reports a successful fetch, closing the circuit.

func (*CircuitBreaker) State 

func (cb *CircuitBreaker) State(domain string) breakerState

State returns a domain's current circuit state. For tests and diagnostics.

type Deduper 

type Deduper interface {
	// MarkIfUnseen atomically claims a URL, reporting whether it was new.
	MarkIfUnseen(rawURL string) bool

	// IsSeen reports whether a URL has been claimed.
	IsSeen(rawURL string) bool

	// Count returns how many unique URLs have been claimed.
	Count() int

	// Export and Import carry state across a checkpoint.
	Export() []string
	Import(hashes []string)

	// Reset clears all state.
	Reset()
}

Deduper is the deduplication strategy the engine uses to decide whether a URL has already been claimed.

Two implementations, with genuinely different trades:

  • Deduplicator keeps every URL hash in a map. Exact, and costs roughly 40 bytes per URL, so a billion-URL crawl needs tens of gigabytes.
  • BloomDeduplicator keeps a Bloom filter and nothing else. Roughly 1.8 bytes per URL at a 1% false-positive rate, and *lossy*: a false positive means a URL is treated as already-seen and never crawled.

type Deduplicator 

type Deduplicator struct {
	// contains filtered or unexported fields
}

Deduplicator tracks visited URLs to avoid re-crawling.

func NewDeduplicator 

func NewDeduplicator(estimatedCapacity int) *Deduplicator

NewDeduplicator creates a new Deduplicator with the given estimated capacity.

func (*Deduplicator) Count 

func (d *Deduplicator) Count() int

Count returns the number of unique URLs seen.

func (*Deduplicator) Export 

func (d *Deduplicator) Export() []string

Export returns all seen URL hashes (for checkpoint serialization).

func (*Deduplicator) Import 

func (d *Deduplicator) Import(hashes []string)

Import loads URL hashes (for checkpoint restore).

func (*Deduplicator) IsSeen 

func (d *Deduplicator) IsSeen(rawURL string) bool

IsSeen returns true if the URL (after canonicalization) has been seen before.

func (*Deduplicator) MarkIfUnseen 

func (d *Deduplicator) MarkIfUnseen(rawURL string) bool

MarkIfUnseen atomically marks a URL as seen and reports whether it was new.

This is the method the crawl path must use. Calling IsSeen followed by MarkSeen is a check-then-act race: two workers extracting the same link from two different pages can both observe "unseen" between the read lock and the write lock, and both enqueue it. The result is duplicate fetches, duplicate items, and a request budget spent twice on the same page.

func (*Deduplicator) MarkSeen 

func (d *Deduplicator) MarkSeen(rawURL string)

MarkSeen marks a URL as seen.

func (*Deduplicator) Reset 

func (d *Deduplicator) Reset()

Reset clears all seen URLs.

type DomainGate 

type DomainGate interface {
	// Ready reports whether the domain could be dispatched now, without
	// consuming anything. Used while scanning candidates, so that inspecting a
	// request the worker then declines does not spend its budget.
	Ready(domain string) bool

	// Claim consumes the domain's allowance and reports whether it succeeded.
	// Called only for the request actually being taken.
	Claim(domain string) bool

	// Delay reports how long until the domain is ready.
	Delay(domain string) time.Duration
}

DomainGate decides whether a request to a given domain may be dispatched now. Throttler implements it; a nil gate means everything is always ready.

type DomainStats 

type DomainStats struct {
	Requests  int64
	Responses int64
	Errors    int64
	LastFetch time.Time
}

DomainStats tracks per-domain statistics.

type Engine 

type Engine struct {
	// contains filtered or unexported fields
}

Engine is the core crawler orchestrator.

func New 

func New(cfg *config.Config, logger *slog.Logger) *Engine

New creates a new Engine with the given configuration.

func (*Engine) AddRequest 

func (e *Engine) AddRequest(req *types.Request) error

AddRequest adds a request to the crawl frontier.

func (*Engine) AddSeed 

func (e *Engine) AddSeed(rawURL string) error

AddSeed adds a seed URL to the crawl frontier.

func (*Engine) ClearCheckpoint 

func (e *Engine) ClearCheckpoint() error

ClearCheckpoint removes the checkpoint file. Called after a crawl finishes normally, so the next run does not resume work that is already done.

func (*Engine) GetState 

func (e *Engine) GetState() State

State returns the current engine state.

func (*Engine) HasCheckpoint 

func (e *Engine) HasCheckpoint() bool

HasCheckpoint reports whether a checkpoint from a previous run is available.

func (*Engine) OnResponse 

func (e *Engine) OnResponse(name string, cb ResponseCallback)

OnResponse registers a named callback for response processing.

func (*Engine) Pause 

func (e *Engine) Pause()

Pause pauses the engine.

func (*Engine) ResultsChan 

func (e *Engine) ResultsChan() <-chan *types.Item

ResultsChan returns a channel that receives a copy of every scraped item.

Each call registers an independent subscriber, so multiple consumers each see the full stream and none of them competes with storage. Previously this handed back the same channel storeResults was draining, which meant every item went to exactly one of them at random: reading the "results" silently corrupted the output file.

Call this before Start. Subscribers registered mid-crawl only see items scraped after they register. The channel is closed when the crawl finishes.

A slow subscriber applies backpressure to the crawl once its buffer fills, which is deliberate — dropping items to keep a consumer fast is a worse failure than slowing down.

func (*Engine) Resume 

func (e *Engine) Resume()

Resume resumes a paused engine.

func (*Engine) ResumeFromCheckpoint 

func (e *Engine) ResumeFromCheckpoint() error

ResumeFromCheckpoint restores frontier, dedup, and stats from the last checkpoint. Distinct from Resume, which un-pauses a running engine.

Call before Start and before adding seeds. Checkpointing was previously write-only: Save ran on a ticker and Load had no caller outside tests, so the crawler produced checkpoint files that nothing ever read.

Seeds added after a resume are still filtered through the restored dedup set, so re-running the same command with the same seeds does not re-crawl what the previous run already covered — the restored frontier is the remaining work, and the seeds are a no-op unless they are genuinely new.

func (*Engine) SetFetcher 

func (e *Engine) SetFetcher(fetcherType string, f Fetcher)

SetFetcher registers a fetcher for a given type.

func (*Engine) SetMetrics 

func (e *Engine) SetMetrics(m *observability.Metrics)

SetMetrics attaches a Prometheus metrics recorder. Call before Start. Passing nil (or never calling this) disables metric recording.

func (*Engine) SetParser 

func (e *Engine) SetParser(p Parser)

SetParser sets the parser implementation.

func (*Engine) SetPipeline 

func (e *Engine) SetPipeline(p Pipeline)

SetPipeline sets the pipeline implementation.

func (*Engine) SetStorage 

func (e *Engine) SetStorage(s Storage)

SetStorage sets the storage implementation.

func (*Engine) Start 

func (e *Engine) Start() error

Start begins crawling.

func (*Engine) Stats 

func (e *Engine) Stats() *Stats

Stats returns the current crawl statistics.

func (*Engine) Stop 

func (e *Engine) Stop()

Stop gracefully stops the engine.

func (*Engine) Wait 

func (e *Engine) Wait()

Wait blocks until all work is done.

Safe to call more than once: the shutdown body runs under a sync.Once, so a second Wait blocks until the first has finished rather than panicking on a double channel close. Callers reasonably treat Wait as idempotent, and "close of closed channel" is a poor way to find out otherwise.

type Fetcher 

type Fetcher interface {
	Fetch(ctx context.Context, req *types.Request) (*types.Response, error)
	Close() error
}

Fetcher is the interface for all fetcher implementations.

type Frontier 

type Frontier struct {
	// contains filtered or unexported fields
}

Frontier is a thread-safe priority queue of crawl requests.

Waiting is event-driven: a blocked Pop parks on a channel and is woken by the Push that gave it work, or by Close. The previous implementation polled the heap every 50 ms, which put a ~25 ms median floor under every dequeue and burned two lock acquisitions per worker per second while idle.

func NewFrontier 

func NewFrontier() *Frontier

NewFrontier creates a new Frontier.

func (*Frontier) Close 

func (f *Frontier) Close()

Close closes the frontier, unblocking every waiting Pop. Safe to call more than once — the idle monitor and an explicit Stop can both reach it.

func (*Frontier) Drain 

func (f *Frontier) Drain() []*types.Request

Drain returns all remaining requests, removing them from the queue.

func (*Frontier) IsClosed 

func (f *Frontier) IsClosed() bool

IsClosed returns true if the frontier has been closed.

func (*Frontier) IsEmpty 

func (f *Frontier) IsEmpty() bool

IsEmpty returns true if the frontier is empty.

func (*Frontier) Len 

func (f *Frontier) Len() int

Len returns the number of requests in the frontier.

func (*Frontier) Pop 

func (f *Frontier) Pop(ctx context.Context) *types.Request

Pop removes and returns the highest-priority request, blocking until one is available, the frontier is closed, or ctx is cancelled. Returns nil in the latter two cases.

func (*Frontier) PopReady 

func (f *Frontier) PopReady(ctx context.Context, gate DomainGate) *types.Request

PopReady removes and returns the highest-priority request whose domain is ready to be fetched, blocking until one exists, the frontier closes, or ctx is done.

This is what stops a throttled domain from parking a worker. The old scheduler dequeued first and *then* slept out the politeness delay while holding the domain's lock — so one slow domain occupied every worker slot in turn and every other domain starved. Here a worker skips past a throttled domain to whatever else is runnable, and only waits when nothing at all is.

func (*Frontier) Push 

func (f *Frontier) Push(req *types.Request)

Push adds a request to the frontier.

func (*Frontier) RestoreAll 

func (f *Frontier) RestoreAll(reqs []*types.Request)

RestoreAll adds multiple requests back (for checkpoint restore).

func (*Frontier) Snapshot 

func (f *Frontier) Snapshot() []*types.Request

Snapshot returns a copy of all queued requests without removing them. Safe for use during checkpointing while the crawl is running.

func (*Frontier) TryPop 

func (f *Frontier) TryPop() *types.Request

TryPop attempts a non-blocking dequeue. Returns nil if empty.

type Parser 

type Parser interface {
	Parse(resp *types.Response, rules []config.ParseRule) ([]*types.Item, []string, error)
}

Parser is the interface for all parser implementations.

type Pipeline 

type Pipeline interface {
	Process(item *types.Item) (*types.Item, error)
}

Pipeline is the interface for the item processing pipeline.

type ResponseCallback 

type ResponseCallback func(resp *types.Response) ([]*types.Item, []*types.Request, error)

ResponseCallback is a function called when a response is received.

type RobotsManager 

type RobotsManager struct {
	// contains filtered or unexported fields
}

RobotsManager handles robots.txt fetching, parsing, and enforcement.

func NewRobotsManager 

func NewRobotsManager(enabled bool, guard *safety.URLGuard) *RobotsManager

NewRobotsManager creates a new RobotsManager.

guard may be nil, in which case the default policy applies. The robots.txt URL is derived from the crawl target's own host, so it is exactly as untrusted as the page it governs and needs the same guarded client.

func (*RobotsManager) GetCrawlDelay 

func (rm *RobotsManager) GetCrawlDelay(domain string) time.Duration

GetCrawlDelay returns the crawl-delay for a domain, if specified.

func (*RobotsManager) GetSitemaps 

func (rm *RobotsManager) GetSitemaps(domain string) []string

GetSitemaps returns the sitemaps listed in robots.txt for a domain.

func (*RobotsManager) IsAllowed 

func (rm *RobotsManager) IsAllowed(rawURL string) bool

IsAllowed checks if a URL is allowed by the domain's robots.txt.

type ScaleAction 

type ScaleAction int

ScaleAction represents a scaling direction. 

const (
	ScaleNone ScaleAction = iota
	ScaleUp
	ScaleDown
)

type ScaleDecision 

type ScaleDecision struct {
	Current     int
	Desired     int
	Utilization float64
	LoadScore   float64
	QueueSize   int
	Action      ScaleAction
}

ScaleDecision describes the outcome of an Evaluate call.

type Scheduler 

type Scheduler struct {
	// contains filtered or unexported fields
}

Scheduler manages worker goroutines that dequeue from the frontier and dispatch fetches.

func NewScheduler 

func NewScheduler(e *Engine) *Scheduler

NewScheduler creates a new Scheduler.

func (*Scheduler) Pause 

func (s *Scheduler) Pause()

Pause pauses all workers.

The gate channel is swapped under pauseMu rather than reassigned in place: workers read it in a select, and an unsynchronised write both races with that read and can park a worker on the *new* channel, which the matching Resume has already closed past — a permanent missed wakeup.

func (*Scheduler) Resume 

func (s *Scheduler) Resume()

Resume resumes all workers.

func (*Scheduler) Start 

func (s *Scheduler) Start(ctx context.Context)

Start launches the worker pool and idle monitor.

func (*Scheduler) Wait 

func (s *Scheduler) Wait()

Wait blocks until all workers are done.

type State 

type State int32

State represents the engine's current lifecycle state. 

const (
	StateIdle     State = 0
	StateRunning  State = 1
	StatePaused   State = 2
	StateStopping State = 3
	StateStopped  State = 4
)

func (State) String 

func (s State) String() string

type Stats 

type Stats struct {
	RequestsSent    atomic.Int64
	RequestsFailed  atomic.Int64
	ResponsesOK     atomic.Int64
	ResponsesError  atomic.Int64
	ItemsScraped    atomic.Int64
	ItemsDropped    atomic.Int64
	URLsEnqueued    atomic.Int64
	URLsFiltered    atomic.Int64
	BytesDownloaded atomic.Int64
	ActiveWorkers   atomic.Int32
	StartTime       time.Time
	// contains filtered or unexported fields
}

Stats tracks crawl statistics.

func (*Stats) Snapshot 

func (s *Stats) Snapshot() map[string]any

Snapshot returns a copy of stats safe for reading.

type Storage 

type Storage interface {
	Store(items []*types.Item) error
	Close() error
}

Storage is the interface for all storage backends.

type Throttler 

type Throttler struct {
	// contains filtered or unexported fields
}

Throttler enforces politeness per registrable domain.

It replaces a `time.Sleep` performed while holding the domain's mutex, after the worker had already dequeued a request. That arrangement had two costs. The sleep occupied a worker slot, so with `politeness_delay: 1s` and 100 workers on one domain the hundredth worker blocked for ~100 seconds doing nothing. And because the delay was applied after dequeue rather than at scheduling time, a single slow domain parked the entire pool and starved every *other* domain — effective concurrency on a single-domain crawl collapsed to one.

Two things change here. Waiting is done by a rate.Limiter rather than a sleep under a lock, so the domain's own mutex is never held across the wait and concurrent callers for different domains never contend. And Allow lets a worker ask whether a domain is ready *before* committing to it, so the scheduler can skip a throttled domain instead of parking on it.

func NewThrottler 

func NewThrottler(delay time.Duration, maxSlots int) *Throttler

NewThrottler builds a throttler enforcing one request per delay per domain. A delay of zero or less disables throttling entirely.

func (*Throttler) Allow 

func (t *Throttler) Allow(domain string) bool

Allow reports whether a request to domain may proceed immediately, consuming a token if so. It never blocks, so a worker can use it to decide whether to take a request or leave it for later without ever parking on a slow domain.

func (*Throttler) Claim 

func (t *Throttler) Claim(domain string) bool

Claim consumes domain's allowance for the request actually being taken.

func (*Throttler) Delay 

func (t *Throttler) Delay(domain string) time.Duration

Delay reports how long until domain is ready. It satisfies DomainGate.

func (*Throttler) Enabled 

func (t *Throttler) Enabled() bool

Enabled reports whether the throttler will delay anything.

func (*Throttler) Len 

func (t *Throttler) Len() int

Len returns the number of tracked domains. Used by tests to assert eviction.

func (*Throttler) Ready 

func (t *Throttler) Ready(domain string) bool

Ready reports whether domain could be dispatched now, without consuming its allowance. The scan in Frontier.PopReady inspects many candidates and takes at most one, so checking must not spend anything.

func (*Throttler) Reserve 

func (t *Throttler) Reserve(domain string) time.Duration

Reserve returns how long a request to domain must wait, without consuming a token. Used to decide how long to park when nothing is runnable, and to scan candidates the worker may not end up taking.

Deliberately not built on ReserveN+Cancel: rate.Reservation.Cancel restores the token only when the reservation's act time is still in the future, so cancelling a zero-delay reservation is a no-op and the token stays spent. That made the "non-consuming" probe consume one token per candidate inspected.

func (*Throttler) Wait 

func (t *Throttler) Wait(ctx context.Context, domain string) error

Wait blocks until a request to domain may proceed, or ctx is done.

This is the fallback for when the scheduler has already committed to a request — the frontier held nothing else runnable — so someone has to wait. Unlike the old sleep, it holds no lock while waiting, so other domains are unaffected.

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.