distribute

package
v0.2.0 Latest Latest
Warning

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

 Go to latest Published: Jul 2, 2026 License: MIT Imports: 10 Imported by: 0

Documentation

Overview

Package distribute is the multi-partition layer (doc 12): the partition map that says which partition owns a HostKey, the router that sends a discovery to that owner, the control plane that owns the map and its epoch, and the rebalance that moves a host slice from one partition to another by shipping a .meguri file. A host lives entirely on one partition (D2), so every question the fleet has a shared answer to reduces to "which partition owns this HostKey", and the map answers it with jump consistent hashing and no directory.

Index

Constants

View Source 
const DefaultFailThreshold = 3

DefaultFailThreshold is the number of consecutive missed heartbeats that moves a partition from Degraded to Failed. A handful of misses rides out a transient stall; a sustained run is a real loss (doc 12, section 5).

Variables

This section is empty.

Functions

func DecodeBatch 

func DecodeBatch(body []byte) ([]m.Discovery, bool)

DecodeBatch reverses EncodeBatch, returning the rows in the encoder's sorted order and whether the body was well-formed. A truncated or malformed body reports not-ok rather than returning partial rows.

func EncodeBatch 

func EncodeBatch(batch []m.Discovery) []byte

EncodeBatch serializes a discovery batch to the columnar delta+FSST body. An empty batch is the empty body. The rows are sorted by URLKey first, so the HostKey column is ascending and codes as plain deltas and the URL spans land in key order.

func MovingHosts 

func MovingHosts(hosts []m.HostRecord, self PartitionID, nm *Map) []uint64

MovingHosts returns the HostKeys this partition holds under the old map but no longer owns under the new one, by scanning the small host table once and testing each key against the new map (doc 12, section 3). These are the hosts whose rows the rebalance slices out and ships; the large URL table is never scanned to find them.

func RecoverInFlight 

func RecoverInFlight(p *format.Partition) int

RecoverInFlight resets every URL stuck InFlight back to Scheduled and returns how many it reset. A URL InFlight at a failure either was never fetched or its outcome was lost, and re-fetching is the safe choice: a redundant polite fetch wastes one request, a lost fetch loses a crawl, so recovery always errs toward the redundant fetch (doc 12, section 5; doc 04, the recovery section).

func Redistribute 

func Redistribute(src *format.Partition, self PartitionID, nm *Map) (ship map[PartitionID]*format.Partition, keep *format.Partition)

Redistribute computes a rebalance for a source partition whose ownership map changed: it groups the hosts that moved by their new owner, slices each destination's hosts out as its own .meguri-ready partition stamped with the destination id, and returns those ship slices plus the keep partition the source retains. A host moves whole or not at all, because politeness is per-host (D2), so the slices and the keep partition never split a host.

The move is a file operation: each slice is a normal partition the source encodes and ships, and the destination merges it into its live store (doc 11). Because the URL table is sorted by HostKey, slicing a moving host is a contiguous range, so the redistribution cost is the moved bytes over the bandwidth wall, not a row-by-row migration (doc 12, sections 3 and 8).

func SliceChecksum 

func SliceChecksum(slice []byte) uint32

SliceChecksum is the CRC32C over an encoded partition slice, the integrity tag the source computes before shipping and the destination echoes in its ack. A mismatch means the bytes that arrived are not the bytes that left, so the source must not delete its copy.

Types

type BacklogSource 

type BacklogSource interface {
	Backlog() []int
}

BacklogSource reports the pending-work depth of every partition, the signal the loop scales on. A fleet binds the partitions' live pending counts; a test binds a fixed slice. The slice is indexed by PartitionID, so its length is the current partition count the loop sees.

type Control 

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

Control is the thin control plane: it owns the partition map and partition health and nothing else, and it is never on the data path (doc 12, section 2). Every change bumps the epoch, the one number that orders changes and guards against split-brain, and a partition learns of a change by comparing the epoch it sees on its heartbeat to its cached one, then pulling the new map. If the control plane is down the fleet keeps crawling against its last-known map; only the four control operations (add, remove, pin, fail over) wait on it.

func NewControl 

func NewControl() *Control

NewControl starts a control plane with a single partition and no replicas, the smallest valid fleet, which a caller grows with AddPartition.

func (*Control) AddPartition 

func (c *Control) AddPartition(address string) PartitionID

AddPartition appends a partition at the high end, the only growth jump hashing supports, bumps the count and epoch, and returns the new id. By minimal movement about 1/(n+1) of hosts now map to the new partition and none move between existing ones (doc 12, section 1).

func (*Control) Backlog 

func (c *Control) Backlog() []int

Backlog reports the latest per-partition pending depth the partitions sent on their heartbeats, indexed by PartitionID for the current partition count, so the control plane itself satisfies BacklogSource and the elasticity loop scales on the live fleet signal rather than a hand-fed slice (doc 12, section 7: the backlog rides the heartbeat the control plane already pulls). A partition that has not beat yet reads as zero, the safe default that never triggers a scale-up. A stale report for an id beyond the current count (a partition since removed) is dropped, so the slice length always matches NumPartitions, which is what the loop divides by.

func (*Control) Epoch 

func (c *Control) Epoch() uint64

Epoch returns the current map epoch, the value a heartbeat compares against to decide whether to fetch.

func (*Control) FailMachine 

func (c *Control) FailMachine(id MachineID) []PartitionID

FailMachine removes a lost machine from the fleet and promotes every partition it held the primary for to that partition's highest-ranked surviving replica, which is exactly the next machine in the rendezvous preference list once the dead one is gone. It recomputes placement over the survivors in one pass, bumps the epoch once, and returns the partitions whose primary moved. Partitions that merely had the machine as a replica get a fresh replica from the same pass. A promotion costs nothing beyond loading the replica's recovered state, because the replica is already a partition caught up to the streamed tail (section 4).

func (*Control) FetchMap 

func (c *Control) FetchMap() *Map

FetchMap returns a snapshot of the current map for a router to cache. It is a clone, so the caller can swap it in without sharing the control plane's slices.

func (*Control) Heartbeat 

func (c *Control) Heartbeat(id PartitionID)

Heartbeat records a live beat from a partition. It clears the partition's miss counter and, if the partition had been marked Degraded or Failed, restores it to Alive and bumps the epoch. This is the crash-and-restart path: a process that recovers locally and beats again reclaims its own partition with no ownership change.

func (*Control) HeartbeatLoad 

func (c *Control) HeartbeatLoad(id PartitionID, backlog int)

HeartbeatLoad records a live beat that also carries the partition's current pending-work depth, the backlog signal the elasticity loop scales on. It does the same liveness restore Heartbeat does and stores the reported depth so Control.Backlog can serve the latest value per partition. This is the live binding of the elasticity loop to the fleet (doc 12, section 7): the control plane already hears from every partition, so the backlog rides the beat it already pulls rather than a separate channel.

func (*Control) MissHeartbeat 

func (c *Control) MissHeartbeat(id PartitionID) HealthState

MissHeartbeat records one missed beat from a partition and returns the partition's resulting health. A first miss degrades the partition, riding out a transient stall; failThreshold consecutive misses fail it. It does not move ownership on its own: detection (the health verdict) and action (FailMachine) are kept separate so the control plane decides to fail over only on a confirmed machine loss, not on a slow heartbeat.

func (*Control) NumPartitions 

func (c *Control) NumPartitions() int

NumPartitions returns the current partition count, the value the elasticity loop reads to decide whether it may still grow or shrink the fleet.

func (*Control) Pin 

func (c *Control) Pin(hostKey uint64, to PartitionID)

Pin overrides the jump hash for one HostKey, routing it to a chosen partition regardless of the hash. It is how a middle drain or a hot-host isolation places specific hosts (doc 12, sections 1 and 7).

func (*Control) Primary 

func (c *Control) Primary(id PartitionID) (MachineID, bool)

Primary returns the machine currently running a partition, and false if the partition is unknown.

func (*Control) RemovePartition 

func (c *Control) RemovePartition() (PartitionID, bool)

RemovePartition drops the highest-numbered partition, the only removal jump hashing supports cheaply; its hosts remap back across the survivors. It returns the removed id and true, or false if only one partition remains.

func (*Control) SetHealth 

func (c *Control) SetHealth(id PartitionID, h HealthState)

SetHealth records a partition's health learned from heartbeats and bumps the epoch, so a failover that changes ownership is ordered like any other change.

func (*Control) SetMachines 

func (c *Control) SetMachines(machines []Machine)

SetMachines records the fleet's machines and recomputes replica placement, the rendezvous mapping from partitions to the machines that hold their copies.

func (*Control) SetReplicas 

func (c *Control) SetReplicas(n int)

SetReplicas sets the fleet-wide replication factor and recomputes every partition's replica set under the new N.

type Elasticity 

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

Elasticity is the operator-facing control loop that turns a backlog signal into scale decisions (doc 12, section 7). The mechanisms it drives already exist: a partition is added at the high end or removed from it with an epoch bump (AddPartition/RemovePartition), the override table pins a draining host, and the rebalance ships the moved host slices once the map changes. What was missing was the loop that watches the backlog, decides, and drives those mechanisms with enough hysteresis that it does not flap. Elasticity is that loop.

It is deliberately clock-free: a Tick is one control interval the caller paces, so the loop is deterministic and testable, and the cooldown and breach windows are counted in ticks rather than wall time. The control plane stays off the data path; the loop only changes the map, and the partitions react to the new epoch on their next heartbeat, slicing and shipping the moved hosts themselves.

func NewElasticity 

func NewElasticity(ctl *Control, src BacklogSource, cfg ElasticityConfig) *Elasticity

NewElasticity builds the loop over a control plane and a backlog source. A zero-valued config is filled with working defaults so a caller need only set the water marks it cares about.

func (*Elasticity) Tick 

func (e *Elasticity) Tick() Scale

Tick reads the backlog once and acts at most once: it adds a partition when the mean per-partition backlog has sat above HighWater for Breaches consecutive ticks and the fleet is below MaxPartitions, removes one when the mean has sat below LowWater for Breaches ticks and the fleet is above MinPartitions, and otherwise holds. After any action it enters a cooldown of Cooldown ticks during which it only holds, so a single load swing never triggers a cascade of resizes. It returns the decision so an operator or a test can watch the loop.

type ElasticityConfig 

type ElasticityConfig struct {
	HighWater     int // mean per-partition backlog above which the fleet is under-provisioned
	LowWater      int // mean per-partition backlog below which the fleet is over-provisioned
	MinPartitions int // never remove below this many
	MaxPartitions int // never add above this many
	Breaches      int // consecutive ticks a water mark must hold before the loop acts
	Cooldown      int // ticks to wait after an action before acting again
}

ElasticityConfig sets the loop's thresholds and hysteresis. The water marks are per-partition backlog: the loop scales on the mean depth across partitions, so a fleet that is evenly under water adds capacity and one that is evenly drained gives it back. The band between LowWater and HighWater is the hold zone that keeps a steady fleet steady.

type HealthState 

type HealthState uint8

HealthState is what the control plane learns about a partition from its heartbeats. Draining is the pre-removal state that stops new hosts routing to a partition while its state ships off (doc 12, section 7). 

const (
	Alive HealthState = iota
	Degraded
	Failed
	Draining
)

type Machine 

type Machine struct {
	ID      MachineID
	Weight  float64
	Address string
}

Machine is a fleet member and its capacity weight. Weight biases the rendezvous placement so a machine with twice the capacity wins the primary slot for about twice as many partitions. Address is where the machine accepts a partition's Discovery messages, carried so a promoted replica can publish the new owner's address into the map (doc 12, section 5).

type MachineID 

type MachineID uint32

MachineID names a physical machine in the fleet. A machine hosts one or more partitions; a heavier machine hosts more, so heterogeneity lives in the placement layer, not the map (doc 12, section 7).

type Map 

type Map struct {
	Epoch         uint64                 // bumped on every change, the change order
	NumPartitions int                    // the jump-hash bucket count
	Overrides     map[uint64]PartitionID // pinned HostKeys, usually empty
	Weights       []uint16               // per-partition weight for placement (section 7)
	Replicas      int                    // N, the replication factor (section 4)
	Partitions    []PartitionMeta        // id, address, HostKey range, health
}

Map is the entire partition map: a count, an epoch, a small override table, the replication factor, and a per-partition row. There is no HostKey directory; Owner recomputes ownership from the count every call, so the map a partition caches is a few kilobytes even for a fleet of thousands.

func (*Map) Clone 

func (m *Map) Clone() *Map

Clone returns a deep copy a router can swap in atomically without sharing the override or partition slices with the control plane that produced it.

func (*Map) Owner 

func (m *Map) Owner(hostKey uint64) PartitionID

Owner returns the partition that owns a HostKey under this map. It is the only place the partition count and the overrides are consulted, a pure function of the HostKey and the cached map: an override pin checked first, then the jump hash over the partition count.

type MoveAck 

type MoveAck struct {
	Dest      PartitionID
	Epoch     uint64
	Checksum  uint32
	Committed bool
}

MoveAck is the destination's acknowledgement of a shipped slice. The source deletes the moved hosts from its own partition only when the ack confirms the destination committed the slice durably (Committed), at the map epoch the move was computed for (Epoch), with the bytes intact (Checksum matches what the source shipped). Any mismatch leaves the hosts on the source.

type MoveTransport 

type MoveTransport interface {
	Ship(dest PartitionID, slice []byte, epoch uint64, checksum uint32) (MoveAck, error)
}

MoveTransport ships an encoded partition slice to its new owner and returns the owner's acknowledgement. The in-process implementation merges the slice into a local destination store and acks; the cross-machine implementation sends the bytes over the wire and waits for the remote ack. The source treats both the same way, the way the discovery and replication seams have one local and one remote implementation (doc 04, the transport binding). A transport error is not fatal to the rebalance: the source keeps that destination's hosts and the move retries next round.

type PartitionID 

type PartitionID uint32

PartitionID names a partition, the jump-hash bucket a HostKey maps to. It is the uint32 the .meguri header stamps as partition_id (doc 10).

func Handoff 

func Handoff(src *format.Partition, self PartitionID, nm *Map, epoch uint64, t MoveTransport) (kept *format.Partition, moved []PartitionID, err error)

Handoff ships every moving host slice to its new owner under the new map and drops from the source only the hosts whose destination acknowledged a durable, epoch-matching, checksum-matching commit. It returns the kept partition the source retains (still owning every host whose move did not complete) and the sorted list of destinations that took their slice.

The contract is the safe-move invariant: a host leaves the source only after its new owner durably holds it, so a crash or a dropped ack at any point loses no host. A destination that errors, fails to commit, or returns a mismatched epoch or checksum keeps its hosts on the source, where they are still served and still shipped next round. An encode failure on a slice is the one fatal error, since it means the source could not produce the bytes to ship at all.

type PartitionMeta 

type PartitionMeta struct {
	ID        PartitionID
	Address   string // where to send this partition's Discovery messages
	Health    HealthState
	HostKeyLo uint64
	HostKeyHi uint64
	Primary   MachineID     // the machine running this partition (0 when unplaced)
	Replicas  []PartitionID // the N-1 replicas for this partition's hosts
}

PartitionMeta is the per-partition row the control plane keeps for routing and health. The HostKey range mirrors what the partition's .meguri header records (doc 10), kept here so the manifest and the routers agree on the bounds.

type Replica 

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

Replica is a partition's standby copy: a snapshot loaded once, then the log tail streamed onto it in LSN order. At every moment it is a partition that has recovered up to the last tail entry it applied, so promotion (section 5) is just materializing it and resetting in-flight work, with no new mechanism: the snapshot ship is the rebalance file ship and the tail stream is the recovery replay (doc 12, section 4).

func NewReplica 

func NewReplica(snap *format.Partition, snapLSN uint64) *Replica

NewReplica loads a shipped snapshot as the replica's base state. snapLSN is the frontier LSN the snapshot was consistent as of (the store's Checkpoint cut, doc 11), so the replica only applies tail entries past it.

func (*Replica) AppliedLSN 

func (r *Replica) AppliedLSN() uint64

AppliedLSN is the highest LSN the replica has folded in, the point its state is recovered to.

func (*Replica) Apply 

func (r *Replica) Apply(e TailEntry)

Apply folds one tail entry into the replica. An entry at or below the applied LSN is a redelivery the replica already holds, so it is ignored; this is what lets the replication transport be at-least-once with no commit protocol, the same economy the discovery transport relies on (doc 12, sections 4 and 6).

func (*Replica) Lag 

func (r *Replica) Lag(primaryLSN uint64) uint64

Lag is how far behind the primary the replica is: the LSNs the primary has written but not yet streamed. It is the bounded staleness section 4 names, the few writes a promoted replica re-crawls, which cost a redundant polite fetch rather than a lost URL.

func (*Replica) Partition 

func (r *Replica) Partition() *format.Partition

Partition materializes the replica's current state as a sorted format.Partition, the same shape a checkpoint produces, so a promoted replica writes a normal .meguri file indistinguishable from one the failed primary would have written.

func (*Replica) Promote 

func (r *Replica) Promote() *format.Partition

Promote turns the replica into the partition it will run as: it materializes the recovered state and resets in-flight work conservatively (section 5). The returned partition is ready to load into a live store and resume crawling.

func (*Replica) Stream 

func (r *Replica) Stream(entries []TailEntry)

Stream applies a batch of tail entries in LSN order. The caller may hand them in any order; Replica sorts by LSN so an out-of-order or batched delivery still lands last-writer-wins.

type Router 

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

Router maps a discovery's HostKey to the partition that owns it and ships the remote ones over the transport (doc 12, section 6, and doc 04). It caches a map snapshot and swaps in a newer one on a heartbeat, so routing is a lock-free map load plus the jump-hash arithmetic, with no per-discovery call to the control plane.

The destination batcher is persistent, not per-call: it accumulates one partition's links across many outcomes and flushes a destination only when its batch fills, so links from a page's outcome and the next page's outcome to the same owner coalesce into one message instead of one per outcome. The engine calls Flush at the window edge (when it would otherwise idle) to ship the partials the fill threshold left behind.

func NewRouter 

func NewRouter(self PartitionID, init *Map, tr Transport, batchSize int) *Router

NewRouter builds a router for one partition over an initial map and transport. batchSize is how many discoveries accumulate for a destination before the router ships them as one message; a page's links to one partition coalesce into far fewer messages than links.

func (*Router) Drain 

func (r *Router) Drain() []m.Discovery

Drain reads everything the transport has queued for this partition and returns it as one slice, the inbound discoveries the caller dedups and schedules. It is the receiver half of the transport: at-least-once means a discovery may appear more than once across drains, which the seen-set absorbs.

func (*Router) Flush 

func (r *Router) Flush() error

Flush ships every destination batch the router has accumulated below the per-destination fill size, the across-outcome counterpart to the per-fill flush inside add. The engine calls it when it would otherwise idle, so a partial batch the busy path left behind never waits past the point the partition quiesces. A send error is returned for the caller to retry; the discovery is rediscoverable either way.

func (*Router) Local 

func (r *Router) Local(d m.Discovery) bool

Local reports whether a discovery's host is owned by this partition, so the caller feeds it straight to the local dedup instead of the transport.

func (*Router) Map 

func (r *Router) Map() *Map

Map returns the router's current cached map snapshot.

func (*Router) Owner 

func (r *Router) Owner(hostKey uint64) PartitionID

Owner returns the partition that owns a HostKey under the cached map. 

func (r *Router) RouteLinks(links []m.Discovery) (local []m.Discovery, err error)

RouteLinks classifies one outcome's out-links: it returns the links whose hosts this partition owns for the caller's own intake, and adds every remote link to its owner's persistent batch, which ships as one message when it fills (doc 12, section 6). The batch outlives the call, so links from this outcome coalesce with links from later outcomes to the same owner; the engine calls Flush at the window edge to ship whatever stays below the fill threshold. A stale map only ever costs an extra hop: a link routed to a partition that no longer owns the host re-routes onward, and the receiver's seen-set dedups any redelivery, so RouteLinks never needs an exactly-correct map.

func (*Router) SwapMap 

func (r *Router) SwapMap(next *Map) bool

SwapMap installs a newer map, the heartbeat-pull convergence: the control plane bumps the epoch on a change, the partition fetches the new map, and the router swaps it in for its next routing decision. A stale or equal epoch is ignored so an out-of-order fetch never moves the router backward. It reports whether the swap happened.

type Scale 

type Scale uint8

Scale is one tick's decision. 

const (
	ScaleHold Scale = iota // load is in band, or hysteresis is not yet satisfied
	ScaleUp                // a partition was added
	ScaleDown              // a partition was removed
)

type SignalRouter 

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

SignalRouter splits an import bundle by owning partition and ships each partition its own slice, then on the receive side applies inbound bundles to a sink under an epoch guard (doc 12, D16). It caches a map snapshot and swaps a newer one in on a heartbeat, the same lock-free read as the discovery router.

func NewSignalRouter 

func NewSignalRouter(self PartitionID, init *Map, tr SignalTransport) *SignalRouter

NewSignalRouter builds a signal router for one partition over an initial map and signal transport.

func (*SignalRouter) Apply 

func (r *SignalRouter) Apply(sink SignalSink) int

Apply drains every inbound bundle and applies the entries to the sink, newest epoch winning. A bundle at or below the highest epoch already applied is dropped whole, so out-of-order delivery never reverts a fresher import. It returns the number of bundles applied, so a caller can tell whether an import landed this drain.

func (*SignalRouter) ApplyLocal 

func (r *SignalRouter) ApplyLocal(local m.Signal, sink SignalSink) bool

ApplyLocal applies the local slice of a bundle this partition routed to itself, the same epoch guard as the inbound path so a self-routed bundle and a received one are imported identically.

func (*SignalRouter) Map 

func (r *SignalRouter) Map() *Map

Map returns the router's current cached map snapshot.

func (*SignalRouter) RouteSignal 

func (r *SignalRouter) RouteSignal(s m.Signal) (local m.Signal, err error)

RouteSignal splits a full import bundle by owning partition: every host and URL entry this partition owns is returned as the local bundle, and every remote entry is grouped by its owner and sent as one bundle per destination. The epoch rides every sub-bundle unchanged so each receiver applies the same version guard. A stale map only misroutes an entry, which the next import corrects, so the split never needs an exactly-correct map.

func (*SignalRouter) SwapMap 

func (r *SignalRouter) SwapMap(next *Map) bool

SwapMap installs a newer map, ignoring a stale or equal epoch, mirroring the discovery router so both move on the same heartbeat.

type SignalSink 

type SignalSink interface {
	ImportURLSignal(m.URLSignal)
	ImportHostSignal(m.HostSignal)
}

SignalSink applies the entries of a bundle a partition owns. The prioritizer implements the URL half through ImportPageRank, and the engine fold writes the host half onto HostRecord.HostScore; a test binds a recording double. The router never calls the blend itself, it only delivers owned entries to the sink, so the policy stays in prioritize and the routing stays here.

type SignalTransport 

type SignalTransport interface {
	// SendSignal delivers a bundle to one partition. The bundle holds only the
	// host and URL entries that partition owns, the per-destination split the
	// router computes.
	SendSignal(to PartitionID, s m.Signal) error
	// RecvSignal returns the next inbound bundle for self and true, or false when
	// nothing is queued.
	RecvSignal(self PartitionID) (m.Signal, bool)
}

SignalTransport carries a tsumugi import bundle from the producer (or the one partition that read the import file) to the partition that owns each host, the signal-side twin of Transport (doc 12, D16). It is one-way and at-least-once for the same reason: a bundle is never depended on and a newer epoch overwrites an older one, so a redelivered or dropped bundle is harmless. The two bindings are an in-process channel for one box and a partitioned log for a fleet.

func NewChannelSignalTransport 

func NewChannelSignalTransport(depth int) SignalTransport

NewChannelSignalTransport builds the in-process signal transport whose per-destination channels buffer up to depth bundles before a SendSignal blocks.

type TailEntry 

type TailEntry struct {
	LSN     uint64
	Kind    TailKind
	URL     m.URLRecord // set for TailPutURL
	Host    m.HostRecord
	URLKey  m.URLKey // set for TailDelURL
	HostKey uint64   // set for TailDelHost
}

TailEntry is one entry of the replicated log tail, carrying the LSN that orders it. The LSN is the store's monotonic per-partition sequence number, so a replica applies entries in LSN order and ignores any it already holds, which makes the stream idempotent and safe to redeliver, the same redo-only, last-writer-wins discipline the store's own recovery uses (doc 11, section 5).

func DelHost 

func DelHost(lsn uint64, hostKey uint64) TailEntry

DelHost builds a tail entry for a host tombstone at the given LSN.

func DelURL 

func DelURL(lsn uint64, key m.URLKey) TailEntry

DelURL builds a tail entry for a URL tombstone at the given LSN.

func PutHost 

func PutHost(lsn uint64, rec m.HostRecord) TailEntry

PutHost builds a tail entry for a host write at the given LSN.

func PutURL 

func PutURL(lsn uint64, rec m.URLRecord) TailEntry

PutURL builds a tail entry for a URL write at the given LSN.

type TailKind 

type TailKind uint8

TailKind tags a replication tail entry, the decoded form of one store log frame (doc 11): a record write or a tombstone. The store ships its snapshot once and then streams these in LSN order; a replica applies them exactly as recovery replays the log, so a replica is a partition recovered up to the tail it has received (doc 12, section 4). 

const (
	TailPutURL  TailKind = iota // a URLRecord write
	TailPutHost                 // a HostRecord write
	TailDelURL                  // a URL tombstone (a key left the partition)
	TailDelHost                 // a host tombstone
)

type Transport 

type Transport interface {
	// Send delivers a batch of discoveries to the partition that owns their
	// hosts. The batch is keyed by destination, so all of one host's discoveries
	// land in the destination's one inbound stream. A full destination blocks the
	// producer, which is the backpressure that signals a rebalance, not a drop.
	Send(to PartitionID, batch []m.Discovery) error
	// Recv returns the next inbound batch for self and true, or false when the
	// transport is drained and closed. A partition's inbound queue is the union
	// of the streams for the hosts it owns.
	Recv(self PartitionID) ([]m.Discovery, bool)
}

Transport carries discoveries from the partition that found a link to the partition that owns the link's host (doc 12, section 6). It is one-way and at-least-once: Send may be retried and the receiver may see a discovery more than once, and idempotency comes from the receiver's seen-set, not the transport, so the transport never needs a commit protocol. The two bindings are an in-process channel for one box and a partitioned log for a fleet; the router does not change a line between them.

func NewChannelTransport 

func NewChannelTransport(depth int) Transport

NewChannelTransport builds the in-process transport whose per-destination channels buffer up to depth batches before a Send blocks the producer. It is the single-box and test binding of Transport; a fleet binds a partitioned log behind the same interface.

func NewWireChannelTransport 

func NewWireChannelTransport(depth int) Transport

NewWireChannelTransport builds the in-process wire transport whose per-destination channels buffer up to depth encoded batches. It is the binding a single-box run uses when it wants the fleet body codec on the path, and the one a test uses to exercise encode and decode end to end.

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.