crawlbroker

Documentation

Overview

Package crawlbroker is the node's gRPC edge to the crawl fleet. It is the only place that speaks the CrawlExchange service: it serves a durable queue of crawl orders to crawler streams and receives ingest batches back, exposing them as the plain ports the inner packages consume. Open starts the server; Close stops it.

Index

• Constants
• func MigrateLegacyStorage(ctx context.Context, source *vault.Vault, target *vault.Vault) error
• func MigrateLegacyStorageWithAdmission(ctx context.Context, source *vault.Vault, target *vault.Vault, ...) error
• type Config
• type ControlRegistry
  • func (r *ControlRegistry) CompleteRun(ctx context.Context, target leaseControlTarget) error
  • func (r *ControlRegistry) Enqueue(workerID string, directive yagocrawlcontract.CrawlControlDirective) bool
  • func (r *ControlRegistry) MaximumActiveRuns() int
  • func (r *ControlRegistry) MaximumRedirects() int
  • func (r *ControlRegistry) ProcessPagesPerSecond() int
  • func (r *ControlRegistry) RestartWorkers() int
  • func (r *ControlRegistry) RuntimePolicy() yagocrawlcontract.CrawlerRuntimePolicy
  • func (r *ControlRegistry) RuntimeSnapshot() CrawlerRuntimeSnapshot
  • func (r *ControlRegistry) SetAutomaticDiscoveryPriority(enabled bool) int
  • func (r *ControlRegistry) SetFetchWorkers(fetchWorkers int) int
  • func (r *ControlRegistry) SetMaximumActiveRuns(maximumActiveRuns int) int
  • func (r *ControlRegistry) SetMaximumRedirects(maximum int) int
  • func (r *ControlRegistry) SetProcessPagesPerSecond(pagesPerSecond int) int
  • func (r *ControlRegistry) SetRuntimePolicy(policy yagocrawlcontract.CrawlerRuntimePolicy) bool
  • func (r *ControlRegistry) SetStoragePressurePolicy(policy yagocrawlcontract.StoragePressurePolicy)
  • func (r *ControlRegistry) StoragePressurePolicy() yagocrawlcontract.StoragePressurePolicy
• type CrawlBroker
  • func Open(cfg Config, storage *vault.Vault, progress ProgressSink) (*CrawlBroker, error)
  • func (b *CrawlBroker) Close()
  • func (b *CrawlBroker) SetURLDenylistSource(source CrawlURLDenylistSource)
• type CrawlURLDenylistSource
• type CrawlerRuntimeSnapshot
• type DurableOrderQueue
  • func (q *DurableOrderQueue) Depth(ctx context.Context) (QueueDepth, error)
  • func (q *DurableOrderQueue) Publish(ctx context.Context, order yagocrawlcontract.CrawlOrder) error
  • func (q *DurableOrderQueue) PublishOnce(ctx context.Context, key string, order yagocrawlcontract.CrawlOrder) (bool, error)
  • func (q *DurableOrderQueue) SetAutomaticDiscoveryPriority(enabled bool)
• type GrowthAdmission
• type IngestReceiver
  • func (r *IngestReceiver) Outstanding() int
  • func (r *IngestReceiver) Receive() <-chan crawlresults.IngestDelivery
• type ProgressSink
• type QueueDepth
  • func (d QueueDepth) Outstanding() int

Constants

const (

	// DefaultLeaseTTL is how long a streamed order stays leased before a missing
	// heartbeat lets the sweeper reclaim and redeliver it.
	DefaultLeaseTTL = 2 * time.Minute
)

Functions

func MigrateLegacyStorage

func MigrateLegacyStorage(ctx context.Context, source *vault.Vault, target *vault.Vault) error

func MigrateLegacyStorageWithAdmission

func MigrateLegacyStorageWithAdmission(
	ctx context.Context,
	source *vault.Vault,
	target *vault.Vault,
	admission vault.RetainedBucketMigrationAdmission,
) error

Types

type Config

type Config struct {
	ListenAddr                        string
	LeaseTTL                          time.Duration
	FetchWorkers                      int
	ProcessPagesPerSecond             int
	MaximumRedirects                  int
	MaximumActiveRuns                 int
	DisableAutomaticDiscoveryPriority bool
	StoragePressurePolicy             yagocrawlcontract.StoragePressurePolicy
	RuntimePolicy                     yagocrawlcontract.CrawlerRuntimePolicy
	GrowthAdmission                   GrowthAdmission
}

type ControlRegistry

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

func (*ControlRegistry) CompleteRun

func (r *ControlRegistry) CompleteRun(
	ctx context.Context,
	target leaseControlTarget,
) error

func (*ControlRegistry) Enqueue

func (r *ControlRegistry) Enqueue(
	workerID string,
	directive yagocrawlcontract.CrawlControlDirective,
) bool

Enqueue queues a directive for its current run worker or the supplied worker. A blank worker id is ignored, since there is no heartbeat to carry it.

func (*ControlRegistry) MaximumActiveRuns

func (r *ControlRegistry) MaximumActiveRuns() int

func (*ControlRegistry) MaximumRedirects

func (r *ControlRegistry) MaximumRedirects() int

func (*ControlRegistry) ProcessPagesPerSecond

func (r *ControlRegistry) ProcessPagesPerSecond() int

func (*ControlRegistry) RestartWorkers

func (r *ControlRegistry) RestartWorkers() int

func (*ControlRegistry) RuntimePolicy

func (r *ControlRegistry) RuntimePolicy() yagocrawlcontract.CrawlerRuntimePolicy

func (*ControlRegistry) RuntimeSnapshot

func (r *ControlRegistry) RuntimeSnapshot() CrawlerRuntimeSnapshot

func (*ControlRegistry) SetAutomaticDiscoveryPriority

func (r *ControlRegistry) SetAutomaticDiscoveryPriority(enabled bool) int

func (*ControlRegistry) SetFetchWorkers

func (r *ControlRegistry) SetFetchWorkers(fetchWorkers int) int

func (*ControlRegistry) SetMaximumActiveRuns

func (r *ControlRegistry) SetMaximumActiveRuns(maximumActiveRuns int) int

func (*ControlRegistry) SetMaximumRedirects

func (r *ControlRegistry) SetMaximumRedirects(maximum int) int

func (*ControlRegistry) SetProcessPagesPerSecond

func (r *ControlRegistry) SetProcessPagesPerSecond(pagesPerSecond int) int

func (*ControlRegistry) SetRuntimePolicy

func (r *ControlRegistry) SetRuntimePolicy(
	policy yagocrawlcontract.CrawlerRuntimePolicy,
) bool

func (*ControlRegistry) SetStoragePressurePolicy

func (r *ControlRegistry) SetStoragePressurePolicy(
	policy yagocrawlcontract.StoragePressurePolicy,
)

func (*ControlRegistry) StoragePressurePolicy

func (r *ControlRegistry) StoragePressurePolicy() yagocrawlcontract.StoragePressurePolicy

type CrawlBroker

type CrawlBroker struct {
	Orders  *DurableOrderQueue
	Ingest  *IngestReceiver
	Control *ControlRegistry
	// contains filtered or unexported fields
}

func Open

func Open(cfg Config, storage *vault.Vault, progress ProgressSink) (*CrawlBroker, error)

func (*CrawlBroker) Close

func (b *CrawlBroker) Close()

func (*CrawlBroker) SetURLDenylistSource

func (b *CrawlBroker) SetURLDenylistSource(source CrawlURLDenylistSource)

type CrawlURLDenylistSource

type CrawlURLDenylistSource func() (yagocrawlcontract.CrawlURLDenylist, error)

type CrawlerRuntimeSnapshot

type CrawlerRuntimeSnapshot struct {
	ConnectedCrawlers              int
	ActiveFetches                  int
	ActiveFetchesKnown             bool
	FetchLimitPerCrawler           int
	AggregateFetchCapacity         int
	StorageStatesKnown             bool
	StorageReportedCrawlers        int
	StorageUnreportedCrawlers      int
	StoragePressured               int
	StorageMeasurementsUnavailable int
	MinimumStorageAvailableBytes   uint64
	StoragePressurePolicy          yagocrawlcontract.StoragePressurePolicy
}

type DurableOrderQueue

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

DurableOrderQueue is a FIFO of crawl orders persisted in the node's storage so queued orders survive a node restart and stay claimable across crawler restarts. Orders move from the pending FIFO into a leased state when streamed to a worker; a lease is settled by an ack, requeued by a nak, and reclaimed when it expires without a heartbeat.

func (*DurableOrderQueue) Depth

func (q *DurableOrderQueue) Depth(ctx context.Context) (QueueDepth, error)

Depth counts the crawl order backlog in a read-only transaction, separating the pending FIFO from the leased in-flight orders.

func (*DurableOrderQueue) Publish

func (q *DurableOrderQueue) Publish(ctx context.Context, order yagocrawlcontract.CrawlOrder) error

Publish enqueues a crawl order for delivery without idempotency. It satisfies the crawl dispatch endpoint's order queue port through PublishOnce.

func (*DurableOrderQueue) PublishOnce

func (q *DurableOrderQueue) PublishOnce(
	ctx context.Context,
	key string,
	order yagocrawlcontract.CrawlOrder,
) (bool, error)

PublishOnce enqueues a crawl order for delivery. When key is non-empty and has already been accepted, nothing is enqueued and duplicate is true, so a retried crawl-start request with the same idempotency key does not create a second order. An empty key disables idempotency and always enqueues.

func (*DurableOrderQueue) SetAutomaticDiscoveryPriority

func (q *DurableOrderQueue) SetAutomaticDiscoveryPriority(enabled bool)

type GrowthAdmission

type GrowthAdmission interface {
	CheckGrowth() error
}

type IngestReceiver

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

IngestReceiver hands ingest batches submitted over gRPC to the node's ingest consumer. SubmitIngest blocks on Receive until the consumer takes the delivery, which is the backpressure the crawler observes.

func (*IngestReceiver) Outstanding

func (r *IngestReceiver) Outstanding() int

func (*IngestReceiver) Receive

func (r *IngestReceiver) Receive() <-chan crawlresults.IngestDelivery

type ProgressSink

type ProgressSink interface {
	Record(ctx context.Context, progress yagocrawlcontract.CrawlRunProgress)
	RecordTerminal(
		ctx context.Context,
		orderIdentity []byte,
		progress yagocrawlcontract.CrawlRunProgress,
	) error
	ConfirmTerminalDelivery(ctx context.Context, orderIdentity []byte) error
}

ProgressSink receives crawl run progress reported by workers. The broker forwards decoded reports to it, so the node's run registry stays decoupled from the gRPC contract.

type QueueDepth

type QueueDepth struct {
	Pending int
	Leased  int
}

QueueDepth is the crawl order backlog held by the broker: Pending orders await a worker lease, while Leased orders are in flight with a worker until acked.

func (QueueDepth) Outstanding

func (d QueueDepth) Outstanding() int

Outstanding is the total crawl work the broker holds, whether waiting for a worker or already leased to one, so a single scalar reflects work in progress rather than dropping to zero the moment an order is leased.