mirror

package
v1.14.1 Latest Latest
Warning

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

Go to latest
Published: Nov 5, 2026 License: Apache-2.0 Imports: 27 Imported by: 0

Documentation

Overview

Package mirror replays one chain's ManageEntity and Plays transactions onto another, in order, exactly once.

It exists for the genesis rollover. The new chain is seeded from a snapshot of the old one, and everything the old chain accepts after that snapshot has to reach the new chain too, until every node has moved and the old chain stops receiving writes. Reading the old chain's blocks rather than the API's relay queue is what makes that complete: relay writes, the API's own write endpoints, and plays recorded by nodes still on the old chain all land in old blocks, and none of them pass through a single API code path.

Order

Transactions are replayed in old-chain block order, and each block in its own transaction order. That order is what the old chain applied, so it is the only order guaranteed to be valid: a user's create precedes their update, a grant precedes the app acting under it.

Core validates a ManageEntity against committed state at mempool admission, so a transaction that depends on one still in flight is refused. A refusal therefore does not halt immediately: the mirror waits for everything in flight to commit and retries once. Only a refusal with nothing in flight is final.

Block by block

Each old block's transactions are all committed before the next old block is read, so no new block ever mixes transactions from two old blocks. That keeps the old chain's grouping, and the grouping matters to an indexer: it re-sorts a block's transactions before applying them, keys some rows on block number or block time (announcements, seen-notifications, price history), and checks some rules against block time. Packing several old blocks into one new block would make a from-genesis indexer of the new chain diverge from what the old chain's indexer produced.

The reverse can still happen: an old block's transactions can be split across two new blocks if a proposal starts partway through submitting them. The mirror cannot prevent that, so it detects it and reports every split on /status.

Timestamps

A live ManageEntity takes created_at and updated_at from the block time of the chain it lands on, so a mirrored write is stamped with when it reached the new chain, not when it happened on the old one. While the mirror works through a backlog those times are late; that is harmless because the API indexes those writes from the old chain. It matters from the indexer flip on, which is the reason to flip only once the mirror trails the old chain by a block or two. Plays carry their own timestamps and are unaffected.

Exactly once

Plays carry no deduplication on chain, so a play sent twice counts twice. The mirror keeps a cursor -- the next old-chain position to process -- and only moves it past a transaction once the new chain has committed it. Before submitting anything it asks the new chain whether that hash is already committed, which covers the window between a commit and the cursor write.

Halting

A refused write is never skipped automatically. It may matter, and continuing past it would replay later writes on top of a state the old chain never had. The mirror stops, reports why, and waits for an operator. Once they have looked at it, they can restart with that one transaction's hash in SkipTx; the mirror then passes over exactly that transaction, logs it, and counts it on /status. Nothing else is ever skipped.

One refusal is answered rather than halted on, when an Attester is configured: a track whose audio someone else already holds a claim for, but whose own upload holds that same file. The new chain needs a node to attest the upload before the track; the mirror does that and retries. See attest.go.

Index

Constants

This section is empty.

Variables

View Source
var (
	// ErrRefused is a validation refusal: the target looked at the transaction
	// and said no. It may be a dependency on something still in flight.
	ErrRefused = errors.New("refused by target")
	// ErrFatal means retrying cannot help, e.g. the target is another chain.
	ErrFatal = errors.New("fatal target error")
)

Forward errors are sorted into three kinds, because each needs a different response. See classifyForwardErr.

Functions

func LoadAttestKey

func LoadAttestKey(path string) (*ecdsa.PrivateKey, error)

LoadAttestKey reads a hex private key from a file, tolerating a 0x prefix and surrounding whitespace.

func Main

func Main(args []string) int

Main runs the mirror command and returns the process exit code.

Types

type AttestedTx

type AttestedTx struct {
	Pos             Position `json:"pos"`
	Hash            string   `json:"hash"`
	AttestationHash string   `json:"attestation_hash"`
	UserID          int64    `json:"user_id"`
	Cids            []string `json:"cids"`
}

AttestedTx is a transaction the mirror attested a track's upload for.

type Attester

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

Attester signs content attestations with a registered node's key.

func NewAttester

func NewAttester(key *ecdsa.PrivateKey, uploads UploadSource) *Attester

func (*Attester) Address

func (a *Attester) Address() string

Address is the node address the attestations are signed as.

type CometSource

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

CometSource reads blocks through a node's CometBFT RPC, served at <node>/core/crpc. It uses the CometBFT `block` call rather than core's GetBlock because GetBlock re-sorts a block's transactions (writes before plays, creates first) for display, and the mirror needs the order the chain actually applied.

/core/crpc is only served by nodes with OPENAUDIO_STATE_SYNC_SERVE_SNAPSHOTS, and only for an allowlist of methods that includes status and block.

func NewCometSource

func NewCometSource(nodeURL string, client *http.Client) *CometSource

func (*CometSource) Block

func (s *CometSource) Block(ctx context.Context, height int64) (string, [][]byte, error)

func (*CometSource) ChainID

func (s *CometSource) ChainID(ctx context.Context) (string, error)

func (*CometSource) Tip

func (s *CometSource) Tip(ctx context.Context) (int64, error)

type Config

type Config struct {
	SourceChainID string
	TargetChainID string
	// StartHeight is where a first run begins, when no cursor is saved.
	StartHeight int64
	// MaxInflight bounds transactions submitted but not yet committed.
	MaxInflight int
	// CommitTimeoutBlocks is how long a submitted transaction may stay
	// uncommitted before it is resubmitted once. Core expires a forwarded
	// transaction from its mempool 10 blocks after admission, so this must be
	// longer than that or a resubmit could race the original.
	CommitTimeoutBlocks int64
	// StartupWaitBlocks is how many target blocks to wait before the first
	// submission. A previous run's in-flight transactions have either
	// committed or expired by then, so the committed check is reliable and
	// nothing is submitted twice.
	StartupWaitBlocks int64
	// SkipTx names transactions, by hash, that an operator has looked at and
	// decided not to mirror. Each is passed over when the mirror reaches it.
	SkipTx []string
	// Attester, when set, answers content-auth refusals by attesting the
	// track's own upload. Nil leaves every refusal a halt.
	Attester     *Attester
	PollInterval time.Duration
	// CursorSaveInterval throttles cursor writes while the mirror is busy.
	CursorSaveInterval time.Duration
}

type ConnectTarget

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

ConnectTarget submits to the new chain's validators over ConnectRPC.

Every transaction goes to every validator, in the same order. A node's mempool is FIFO and proposals take from its front, but the gossip between validators is asynchronous, so a transaction submitted to one node alone can reach the next proposer out of order. Submitting to each validator directly gives every proposer the same order. Duplicates from gossip are dropped by the mempool, which keys on the hash.

func NewConnectTarget

func NewConnectTarget(chainID string, nodeURLs []string, client *http.Client) (*ConnectTarget, error)

func (*ConnectTarget) Committed

func (t *ConnectTarget) Committed(ctx context.Context, hash string) (int64, bool, error)

func (*ConnectTarget) Forward

Forward sends tx to every validator at once and waits for all of them, so a transaction costs one round trip to the slowest node rather than the sum of them; a proposal that lands mid-forward would otherwise split a source block across target blocks. Each node still sees transactions in the mirror's order, because the next Forward starts only after this one returns. The error reported is the first one in node order, as before.

func (*ConnectTarget) Status

func (t *ConnectTarget) Status(ctx context.Context) (string, int64, error)

type CursorStore

type CursorStore interface {
	// Load returns the saved cursor, and false when none has been saved.
	Load() (Position, bool, error)
	Save(Position) error
}

CursorStore persists the cursor.

type DryRunReport

type DryRunReport struct {
	From, To int64
	Blocks   int64
	// BlocksWithMirrored is how many blocks hold at least one transaction
	// the mirror would send: each costs about one target block when
	// following block by block, and empty blocks cost nothing.
	BlocksWithMirrored int64
	Transactions       int64
	ManageEntity       int64
	Plays              int64
	NotMirrored        int64
	OnTarget           int64
	HashMismatches     []string
}

DryRunReport is what a read-only scan of source blocks found.

func DryRun

func DryRun(ctx context.Context, source Source, target Target, sourceChainID string, from, n int64) (*DryRunReport, error)

DryRun reads source blocks [from, from+n) and reports what the mirror would do, without submitting anything. It also checks the one assumption the exactly-once guarantee rests on: that a decoded transaction re-marshals to the bytes the source committed, so it keeps its hash on the target. A mismatch means the target would commit it under a different hash and the "already on target" check could not see it.

type FileCursor

type FileCursor struct{ Path string }

FileCursor keeps the cursor in a JSON file, replaced atomically.

func (FileCursor) Load

func (f FileCursor) Load() (Position, bool, error)

func (FileCursor) Save

func (f FileCursor) Save(p Position) error

type HTTPUploads

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

HTTPUploads reads upload rows from a storage node's public GET /uploads/<id>.

func NewHTTPUploads

func NewHTTPUploads(nodeURL string, client *http.Client) *HTTPUploads

func (*HTTPUploads) Upload

func (u *HTTPUploads) Upload(ctx context.Context, id string) (*UploadRecord, error)

type HaltError

type HaltError struct {
	Pos    Position
	Hash   string
	Reason string
	Err    error
}

HaltError stops the mirror. It names the transaction and the reason.

func (*HaltError) Error

func (e *HaltError) Error() string

func (*HaltError) Hint

func (e *HaltError) Hint() string

Hint tells the operator how to continue past this halt, when skipping is an option at all: a halt that names no transaction (wrong chain, say) is not.

func (*HaltError) Unwrap

func (e *HaltError) Unwrap() error

type Mirror

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

func New

func New(cfg Config, source Source, target Target, cursor CursorStore, logger *zap.Logger) *Mirror

func (*Mirror) Init

func (m *Mirror) Init(ctx context.Context) error

Init checks both chains are the ones configured and loads the cursor. It refuses to start against the wrong chain, which is the one mistake that would quietly write the wrong history somewhere.

func (*Mirror) LiveStatus

func (m *Mirror) LiveStatus(ctx context.Context) Status

LiveStatus is Status with the source tip read now. The mirror itself only asks for the tip once its cursor passes the last one it saw, so between those reads Status undercounts the lag by however far the source has moved. If the source does not answer, the snapshot is returned as it is.

func (*Mirror) Run

func (m *Mirror) Run(ctx context.Context) error

Run mirrors until ctx ends or the mirror halts. A halt returns a *HaltError.

func (*Mirror) Status

func (m *Mirror) Status() Status

Status returns a snapshot of the mirror's state.

func (*Mirror) Step

func (m *Mirror) Step(ctx context.Context) (bool, error)

Step does one round of work: settle what has committed, then submit more. It reports whether anything moved, so the caller knows whether to wait.

type Position

type Position struct {
	Height int64 `json:"height"`
	Index  int   `json:"index"`
}

Position is a transaction's place on the source chain: its block, and its index among that block's transactions. The cursor is the next Position to process.

func (Position) String

func (p Position) String() string

type SkippedTx

type SkippedTx struct {
	Pos  Position `json:"pos"`
	Hash string   `json:"hash"`
}

SkippedTx is a transaction passed over because an operator named it.

type Source

type Source interface {
	// ChainID is the chain the source serves.
	ChainID(ctx context.Context) (string, error)
	// Tip is the latest block the source can return.
	Tip(ctx context.Context) (int64, error)
	// Block returns the raw transactions of a block, in block order, and the
	// chain ID in its header.
	Block(ctx context.Context, height int64) (chainID string, txs [][]byte, err error)
}

Source is the old chain, read block by block in block order.

type SplitInfo

type SplitInfo struct {
	SourceHeight  int64   `json:"source_height"`
	TargetHeights []int64 `json:"target_heights"`
}

SplitInfo names a source block that landed in more than one target block.

type Status

type Status struct {
	SourceChainID   string   `json:"source_chain_id"`
	TargetChainID   string   `json:"target_chain_id"`
	SourceTip       int64    `json:"source_tip"`
	Cursor          Position `json:"cursor"`
	LagBlocks       int64    `json:"lag_blocks"`
	Inflight        int      `json:"inflight"`
	Draining        bool     `json:"draining"`
	Submitted       int64    `json:"submitted"`
	Committed       int64    `json:"committed"`
	AlreadyOnTarget int64    `json:"already_on_target"`
	// SplitBlocks counts source blocks whose transactions landed in more than
	// one target block; LastSplit is the most recent.
	SplitBlocks int64      `json:"split_blocks"`
	LastSplit   *SplitInfo `json:"last_split,omitempty"`
	// Skipped lists the operator-skipped transactions the mirror has passed
	// over in this run.
	Skipped []SkippedTx `json:"skipped,omitempty"`
	// Attested lists the attestations the mirror made in this run.
	Attested   []AttestedTx `json:"attested,omitempty"`
	Halted     bool         `json:"halted"`
	HaltReason string       `json:"halt_reason,omitempty"`
	HaltHint   string       `json:"halt_hint,omitempty"`
	LastError  string       `json:"last_error,omitempty"`
	UpdatedAt  time.Time    `json:"updated_at"`
}

Status is what the mirror reports about itself.

type Target

type Target interface {
	// Status returns the target's chain ID and current height.
	Status(ctx context.Context) (chainID string, height int64, err error)
	// Forward submits a transaction to the mempool of every target node.
	Forward(ctx context.Context, tx *corev1.SignedTransaction) error
	// Committed reports whether a transaction hash is in a committed block,
	// and the height of that block.
	Committed(ctx context.Context, hash string) (height int64, ok bool, err error)
}

Target is the new chain.

type UploadRecord

type UploadRecord struct {
	ID          string            `json:"id"`
	OrigFileCID string            `json:"orig_file_cid"`
	Results     map[string]string `json:"results"`
	UserID      sql.NullInt64     `json:"user_id"`
}

UploadRecord is the part of a storage node's upload row the mirror checks.

type UploadSource

type UploadSource interface {
	Upload(ctx context.Context, id string) (*UploadRecord, error)
}

UploadSource looks up an upload by id on the old chain's storage.

Jump to

Keyboard shortcuts

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