Documentation
¶
Overview ¶
Package engine composes the resident frontier with a fetcher worker pool and the distribution router into the staged, single-writer partition loop of doc 04. The frontier is the only writer to its own state: every mutation, whether a routed-discovery intake or an outcome fold, runs on the engine's one run goroutine, while the fetchers run concurrently on a bounded pool sized to the polite-host parallelism. Out-links split at the fold, the frontier's link sink shipping the remote ones to their owners through the router and handing back the local ones for this partition's intake, exactly the doc 04 dataflow.
Two integration directions ride the same frontier. Engine.Run is meguri driving ami: the engine pulls the next polite URL, hands it to the fetcher pool, and folds the outcome back. MeguriSeedSource is the inverse, ami driving meguri: a partition presents the ami.SeedSource shape, Next blocking until a URL is due and polite (doc 13). Both keep the single-writer rule by serializing every frontier call.
Index ¶
- func RouteSink(r *distribute.Router) func([]meguri.Discovery) []meguri.Discovery
- type AmiClient
- type AmiFetcher
- type AmiOption
- type AmiRequest
- type AmiResult
- type BatchFetcher
- type Clock
- type Collection
- type Config
- type Engine
- type LogicalClock
- type MeguriSeedSource
- type Partition
- type Result
- type Seed
- type SeedOption
- type SeedReader
- type Stats
- type WallClock
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func RouteSink ¶
RouteSink adapts a router into the frontier's out-link sink (frontier.With LinkRouter): it ships the remote links to their owners and returns the local subset for this partition's intake. A transport send error is swallowed here on purpose, because the discovery transport is at-least-once and a dropped ship is rediscoverable; the link reappears the next time its source is crawled, and the receiver's seen-set dedups it. Pass it as frontier.WithLinkRouter(engine.RouteSink(router)).
Types ¶
type AmiClient ¶
type AmiClient interface {
Do(ctx context.Context, req AmiRequest) (AmiResult, error)
}
AmiClient is the network seam: the one method the adapter calls to actually retrieve bytes. The production implementation wraps ami's fetch.Fetcher (the live binding); a test serves canned AmiResults so the whole adapter, the request build and the result mapping and the link extraction, runs without a socket. It must be safe for concurrent use: the engine dispatches many requests at once.
type AmiFetcher ¶
type AmiFetcher struct {
// contains filtered or unexported fields
}
AmiFetcher adapts an AmiClient to the engine's fetch.Fetcher SPI. It is the production Fetcher once its client is the live ami binding, and the same type the double drives in the gate. It carries the canonicalization policy and host grouping the out-link extraction keys under, the same ones the seed path uses, so a link and a seed enter the frontier on one keying.
func NewAmiFetcher ¶
func NewAmiFetcher(client AmiClient, opts ...AmiOption) *AmiFetcher
NewAmiFetcher builds the adapter over a client. With the live ami binding as the client it is the production fetcher; with a fixture-serving double it is the in-process gate.
func (*AmiFetcher) Fetch ¶
Fetch runs one request through the ami client and maps its result into the outcome the frontier folds. A transport failure (no HTTP exchange) returns the error, which the engine turns into a retryable no-op; an HTTP error status is a normal outcome carrying that status. A robots fetch returns the raw body for the frontier to parse and never extracts links. A 304 or a digest match is a no-change outcome with the content signals zeroed, as the freshness model reads it. A 2xx extracts, canonicalizes, and keys the body's out-links so the prioritizer can spread the source's cash over them.
type AmiOption ¶
type AmiOption func(*AmiFetcher)
AmiOption configures an AmiFetcher.
func WithAmiCanonPolicy ¶
func WithAmiCanonPolicy(p *dedup.CanonPolicy) AmiOption
WithAmiCanonPolicy sets the link canonicalization policy. The default (nil) is the global default policy, the same one the seed reader uses.
func WithAmiClock ¶
WithAmiClock sets the epoch-hours stamp source for the outcome's FetchedAt and the out-links' ObservedAt, so a replay can stamp deterministically. The default reads the wall clock.
func WithAmiGrouping ¶
func WithAmiGrouping(g meguri.HostGrouping) AmiOption
WithAmiGrouping sets the host grouping out-links key under. The default is the registrable domain, the unit a partition owns and the seed path keys under.
type AmiRequest ¶
type AmiRequest struct {
URL string // the URL to fetch
IfNoneMatch string // prior ETag, for a conditional GET; empty for unconditional
IfModified uint32 // prior Last-Modified epoch-hours, for If-Modified-Since; 0 if none
ResolvedIP [16]byte // the host's cached DNS answer, zero when ami must resolve
Robots bool // this is the host's robots.txt fetch, body returned raw
}
AmiRequest is what the adapter hands an AmiClient: a single URL to fetch with the conditional-GET validators and the resolved IP the politeness bucket accounted for. It is the meguri-shaped view of an ami seed, decoupled from ami's own types so the engine never imports ami; the live binding translates it into ami's request, an offline double answers it from a fixture.
type AmiResult ¶
type AmiResult struct {
Status int // the HTTP status; 0 with Err set means no exchange
Body []byte // the response body, empty on a 304 or a no-body status
ETag string // the response ETag header, empty if none
LastModified string // the response Last-Modified header, raw, empty if none
Unchanged bool // ami's post-fetch digest matched the prior crawl (304-equivalent)
FinalURL string // the URL after ami followed redirects; equals the request URL if none
LatencyMS uint16 // round-trip latency in milliseconds
RetryAfter uint16 // deciseconds from a Retry-After header, 0 if none
CrawlDelay uint16 // deciseconds, a robots Crawl-delay when this was a robots fetch
Err error // transport/DNS/timeout failure, no usable HTTP outcome
}
AmiResult is the meguri-shaped view of ami's fetch.Result: the fields the adapter maps into a meguri.Outcome. It mirrors ami's Result (Status, Header validators, Body, the post-fetch digest match, the redirect-resolved FinalURL, the server IP and timing) without importing it, so the live binding fills it from ami and a double fills it from a fixture. A transport failure that yields no HTTP exchange at all is Err; an HTTP error status is a normal result with that status.
type BatchFetcher ¶
type BatchFetcher struct {
// contains filtered or unexported fields
}
BatchFetcher wraps one concurrency-safe fetch.Fetcher in a bounded worker pool, the shape doc 13 specifies: requests go in, results come out in completion order, not request order, so a slow host never holds a fast one behind it. The pool size is the engine's polite-host parallelism, which is the only thing that bounds the number of fetches in flight (politeness, not the fetcher).
func NewBatchFetcher ¶
func NewBatchFetcher(f fetch.Fetcher, workers int) *BatchFetcher
NewBatchFetcher wraps f in a pool of the given size. A size below one is raised to one so the pool always makes progress.
func (*BatchFetcher) FetchBatch ¶
FetchBatch is the doc-13 literal surface: it fetches a fixed slice of requests through the pool and returns a channel of their outcomes in completion order, closed when the last one lands. A fetch that errors is dropped from the stream rather than surfaced, because FetchBatch is the convenience wrapper for callers that only want outcomes; the engine uses the lower-level stream so it sees the errors and can re-place a failed host.
func (*BatchFetcher) Workers ¶
func (b *BatchFetcher) Workers() int
Workers reports the pool size, the in-flight bound the engine dispatches up to.
type Clock ¶
Clock is the engine's view of time. The run loop reads Now in epoch-seconds to stamp dispatch and outcome folds, and calls SleepUntil when every host is cooling down and nothing is in flight, so the next thing it does is wait for the earliest politeness window to open. Two implementations bind it: a wall clock that actually sleeps for a live crawl, and a logical clock that jumps the instant forward so a corpus replay drains without real waits.
SleepUntil is called only when the pool is empty (no fetch is in flight), so a logical clock can advance the instant with no risk of reordering an in-flight fetch against the clock.
type Collection ¶
type Collection struct {
// contains filtered or unexported fields
}
Collection is the fleet-side reader of a partition manifest: the catalog that binds many .meguri partitions into one frontier (doc 10 section 11). It reads a manifest file once and answers the three fleet questions without opening any partition: which partition owns a host (Route), which partitions have due work (DueParts), and whether the ranges tile the key space cleanly (CoverageGap). A serve process holds it to route a discovered link to its owner, and OpenPartition opens the durable partition a routed entry points at.
func OpenCollection ¶
func OpenCollection(path string) (*Collection, error)
OpenCollection reads and parses a manifest file, the reader half of the fleet catalog. The file is what BuildManifest plus EncodeManifest wrote, verified at both ends by its magic and CRC, so a torn manifest fails loudly here rather than routing to a partition that was never committed.
func (*Collection) Manifest ¶
func (c *Collection) Manifest() *format.Manifest
Manifest returns the parsed catalog, for callers that want the raw entries or the CoverageGap and DueParts pushdowns directly.
func (*Collection) OpenPartition ¶
func (c *Collection) OpenPartition(e format.ManifestEntry, opts store.Options, frOpts ...frontier.Option) (*Partition, error)
OpenPartition opens the durable partition a manifest entry points at, resolving the entry's FileRef as a store directory. It is how a routed lookup turns into a live handle: route a host to its entry, open that entry, drive its frontier.
func (*Collection) Route ¶
func (c *Collection) Route(hk uint64) (format.ManifestEntry, bool)
Route returns the partition entry whose host-key range contains hk, the lookup a router runs to send a discovered link to its owner. ok is false when no partition owns hk, a gap a control plane treats as missing coverage.
type Config ¶
type Config struct {
// Fetcher retrieves one request and returns its outcome; it must be safe for
// concurrent use, since the pool calls it from every worker at once.
Fetcher fetch.Fetcher
// Workers is the polite-host parallelism, the in-flight fetch bound. Zero
// means a small default that keeps a single box busy without oversubscribing
// the fetcher.
Workers int
// Clock is the engine's time source. Zero means a wall clock for a live crawl;
// the corpus gate passes a logical clock so it drains without real waits.
Clock Clock
// Router, when set, is the inbound side of distribution: the engine drains the
// transport for discoveries routed to this partition and folds them into the
// frontier. The outbound side is the frontier's own link sink (WithLinkRouter,
// wired with RouteSink), so a partition both ships and receives cross-partition
// links. Nil is the single-partition case.
Router *distribute.Router
// Signals, when set, is the inbound side of tsumugi signal import (doc 12,
// D16): on each loop the engine drains the signal transport for bundles routed
// to this partition and applies them to the frontier, which blends the imported
// host_score and PageRank into priority. The outbound side is ImportSignal,
// which a partition that reads an import file calls to route a full bundle to
// every owner. Nil leaves a partition on its locally accumulated signals.
Signals *distribute.SignalRouter
// UntilEmpty stops Run when the frontier drains and no inbound work remains,
// the mode the gate and a bounded batch crawl use. False runs until the context
// is cancelled, the mode a live fleet partition uses, waiting on new
// discoveries when its own frontier is momentarily empty.
UntilEmpty bool
}
Config wires an Engine. Only Fetcher is required; the rest take working defaults so a single-box replay run needs nothing more than a fetcher.
type Engine ¶
type Engine struct {
// contains filtered or unexported fields
}
Engine drives one partition's crawl loop. It owns no durable state of its own: the frontier holds the schedule, the fetcher pool holds the in-flight work, and the engine is the single goroutine that moves URLs between them.
func New ¶
New builds an engine over an already-configured frontier. The frontier carries its own policy (politeness, freshness, prioritization) and, for a distributed run, its outbound link sink (frontier.WithLinkRouter wired with RouteSink); the engine adds the loop, the fetcher pool, the clock, and the inbound intake.
func (*Engine) ImportSignal ¶
ImportSignal routes a full tsumugi import bundle to every partition that owns part of it and applies this partition's own slice to the frontier (doc 12, D16). It is the producer-side entry: the partition that reads an import file calls it once, the signal router splits the bundle by owner, ships each remote slice over the signal transport, and the owners drain it on their next loop. A transport error is returned so the caller can retry; the bundle is never depended on, so a dropped import only delays the signal refresh.
func (*Engine) Run ¶
Run drives the crawl loop until the frontier drains (UntilEmpty) or the context is cancelled. It is the single writer to the frontier: it dispatches the next polite URL, places it on the fetcher pool, and folds each returned outcome back, all on this one goroutine, while the pool's workers fetch concurrently. When no host may be fetched and nothing is in flight, it advances the clock to the next politeness window rather than spinning.
type LogicalClock ¶
type LogicalClock struct {
// contains filtered or unexported fields
}
LogicalClock is the replay clock the corpus gate runs: SleepUntil jumps the instant forward instead of waiting, so a slice whose politeness windows span hours drains in milliseconds while the dispatch order stays exactly what the wall clock would produce. The instant only ever moves forward, and only when the pool is empty, so a concurrent fold never sees the clock move under it.
func NewLogicalClock ¶
func NewLogicalClock(start uint32) *LogicalClock
NewLogicalClock starts a logical clock at start (epoch-seconds).
func (*LogicalClock) Now ¶
func (c *LogicalClock) Now() uint32
Now returns the current logical instant.
func (*LogicalClock) SleepUntil ¶
func (c *LogicalClock) SleepUntil(_ context.Context, epochSec uint32)
SleepUntil advances the logical instant to epochSec if it is in the future, the jump that stands in for a wall-clock wait.
type MeguriSeedSource ¶
type MeguriSeedSource struct {
// contains filtered or unexported fields
}
MeguriSeedSource is the inverse integration of Engine.Run: instead of meguri pulling a fetcher, ami pulls meguri, the partition presenting the ami.SeedSource shape (doc 13). Next blocks until a URL is both due and polite and returns its request; ami fetches it and hands the outcome back through Report. Because the frontier is a single-writer structure and ami calls from many goroutines, a mutex serializes every Dispatch and Report, so the partition stays the only writer to its own state even under a concurrent puller.
func NewSeedSource ¶
func NewSeedSource(fr *frontier.Frontier, clk Clock) *MeguriSeedSource
NewSeedSource presents a frontier as an ami seed source. A nil clock means a wall clock; the gate passes a logical clock so a replay drains without waits.
func (*MeguriSeedSource) Next ¶
Next returns the next URL to fetch, blocking until a host is due and polite. It returns ok=false when the frontier is fully drained (no parked host remains) or the context is cancelled, the signal for the puller to stop. While every host is cooling down it advances the clock to the next politeness window rather than busy-waiting, the same wait Engine.Run performs.
func (*MeguriSeedSource) Report ¶
func (s *MeguriSeedSource) Report(o meguri.Outcome)
Report folds a fetched outcome back into the frontier under the same lock Next dispatches under, so the puller's completion handler and its next pull never race on the frontier.
type Partition ¶
type Partition struct {
// contains filtered or unexported fields
}
Partition is the top-level open/close lifecycle of one durable partition: open a directory, get a running frontier backed by the log-structured store, advance it through an Engine, and fold it back to a durable checkpoint on close (doc 11, doc 04). It is the handle the serve command holds and the library entry point a host process embeds: the store is the durable home and the frontier is the resident scheduler, and the lifecycle keeps the two in step so a caller never touches the store and the frontier separately.
The split of responsibilities: store.Open recovers the durable snapshot and log tail, Snapshot hands that state to frontier.Recover as a resident schedule, the Engine advances the frontier, and Checkpoint serializes the advanced frontier back through store.CheckpointFrom. During a session the frontier is the live truth; the store is only read at Open and only written at Checkpoint, so the two never race and the durable form is always a clean cut between engine ticks.
func OpenPartition ¶
OpenPartition opens or recovers the durable partition rooted at dir and returns a handle whose frontier is ready to drive. A fresh directory comes back with an empty frontier to seed; an existing one comes back with its committed schedule recovered. The frontier options (WithStateMachine, WithPrioritization, and the rest) configure the resident scheduler the same way frontier.Recover does directly, so a served partition carries the same policy a one-shot run does.
func (*Partition) Abandon ¶
Abandon closes the store without checkpointing, discarding any frontier advance since the last Checkpoint. It is the shutdown path for a read-only session or a run whose result the caller chooses not to keep; recovery falls back to the last committed checkpoint.
func (*Partition) Checkpoint ¶
Checkpoint folds the live frontier into the store's durable checkpoint: it serializes the frontier to its .meguri form and commits it through the store, which writes the snapshot, rotates the log, and swaps the superblock. After it returns the on-disk partition reflects every advance the engine made, and a later OpenPartition recovers exactly this state.
func (*Partition) Close ¶
Close checkpoints the live frontier and closes the underlying store, the clean shutdown of a served partition. A caller that has already checkpointed and wants to drop the session without re-persisting calls Abandon instead.
func (*Partition) Dir ¶
Dir is the directory the partition is rooted at, the durable home of its log and snapshot.
func (*Partition) Frontier ¶
Frontier returns the resident frontier the lifecycle drives. A caller seeds it, hands it to engine.New, and reads it back; the lifecycle owns persisting it.
func (*Partition) ManifestEntry ¶
func (p *Partition) ManifestEntry(epoch uint32) (format.ManifestEntry, error)
ManifestEntry derives this partition's manifest row from its live frontier: it serializes the frontier to its .meguri form and reads the header and footer the catalog records (range, counts, soonest due, CRC). The FileRef is the partition's own directory, the durable home a Collection reopens it from. A control plane collects one entry per partition into BuildManifest to publish the fleet catalog.
type Result ¶
Result is one finished fetch: the request that produced it, the outcome, and a non-nil Err only when the fetcher could not produce a usable outcome at all (a dead connection, not an HTTP error status, which is a normal outcome). The engine folds a Result back into the frontier on its single run goroutine.
type Seed ¶
type Seed struct {
URL string // the seed URL, canonicalized and keyed by the reader
Base string // optional base for resolving a relative seed; usually empty
Priority float32 // initial importance, carried as the discovery's link weight
PrevETag string // ETag from an earlier crawl, for the first conditional GET
PrevLastModified uint32 // Last-Modified epoch-hours from an earlier crawl
PrevDigest uint64 // content fingerprint from an earlier crawl
Source meguri.DiscoverySource // how it entered; zero means SourceSeed
ObservedAt uint32 // epoch-hours the seed was listed; zero means unknown
}
Seed is one entry of a seed list as ami hands it across the boundary (doc 13): a URL to crawl, an initial priority, the host grouping it keys under, and the validators a prior crawl left so the first fetch can go conditional. It is the producer side; SeedReader turns it into the meguri.Discovery the frontier's idempotent intake ingests and the prior-crawl state Warm stamps on the record.
type SeedOption ¶
type SeedOption func(*SeedReader)
SeedOption configures a SeedReader.
func WithCanonPolicy ¶
func WithCanonPolicy(p *dedup.CanonPolicy) SeedOption
WithCanonPolicy sets the canonicalization policy. The default (nil) is the global default policy: the tracking-parameter deny-list and no host folding.
func WithGrouping ¶
func WithGrouping(g meguri.HostGrouping) SeedOption
WithGrouping sets the host grouping seeds key under. The default is the registrable domain, the same unit a partition owns.
type SeedReader ¶
type SeedReader struct {
// contains filtered or unexported fields
}
SeedReader converts seeds into discoveries with a fixed canonicalization policy and host grouping, the one piece of code that knows both the seed shape and the frontier's keying. It canonicalizes each URL, keys it the same way a crawled out-link is keyed, and copies the seed's priority into the discovery's link weight so a seed and a link enter the prioritizer on the same scale (doc 13).
func NewSeedReader ¶
func NewSeedReader(opts ...SeedOption) *SeedReader
NewSeedReader builds a reader with the default registrable-domain grouping and the global canonicalization policy.
func (*SeedReader) Discovery ¶
func (r *SeedReader) Discovery(s Seed) (meguri.Discovery, bool)
Discovery converts one seed into a discovery, returning false when the URL does not canonicalize to a crawlable http(s) key. The discovery carries the seed's canonical URL inline, its priority as the link weight, and SourceSeed unless the seed names another source.
func (*SeedReader) Ingest ¶
Ingest reads a batch of seeds into a frontier: it converts each to a discovery, folds it through the idempotent intake, and warms the new record with the seed's prior-crawl validators so its first fetch goes conditional. It returns how many seeds entered as new schedulable URLs. Ingest assumes single-threaded access to the frontier (the caller holds the engine's run goroutine or has not started it yet), matching the single-writer rule.
type Stats ¶
Stats is a snapshot of what the engine has moved: URLs dispatched to the pool, outcomes folded back, and fetches that errored without a usable outcome.
Source Files