p2p

Documentation

Overview

Package p2p contains pull-based libp2p synchronization primitives for qi documents. Callers initiate each Pull, so bidirectional convergence requires each side to pull. Resolver authorization must use the authenticated remote libp2p peer, not caller-declared causal replica IDs.

Index

• Constants
• Variables
• func Pull(ctx context.Context, h host.Host, remote peer.AddrInfo, ...) (bool, error)
• func ResolverStreamHandler(resolver SourceResolver) network.StreamHandler
• func Serve(h host.Host, source Source) error
• func ServeResolver(h host.Host, resolver SourceResolver) error
• func StreamHandler(source Source) network.StreamHandler
• type Sink
• type Source
• type SourceResolver
• type SourceResolverFunc
  • func (f SourceResolverFunc) ResolveSource(remote peer.ID, documentID string) (Source, bool)
• type SourceRouter
  • func NewSourceRouter() *SourceRouter
  • func (r *SourceRouter) Register(source Source) error
  • func (r *SourceRouter) ResolveSource(_ peer.ID, documentID string) (Source, bool)
  • func (r *SourceRouter) Unregister(documentID string) bool
• type Store
• type SyncRequest
  • func (r SyncRequest) MarshalJSON() ([]byte, error)
  • func (r *SyncRequest) UnmarshalJSON(data []byte) error
• type SyncResponse

Constants

const MaxFrameBytes int64 = 4 << 20

MaxFrameBytes limits a single JSON-framed sync message.

const ProtocolID = protocol.ID("/qi/sync/2")

ProtocolID is the generation-aware qi sync protocol. Version 2 requests already carry the routing key in Cursor.Identity.DocumentID, so serving multiple documents does not change the protocol ID or JSON framing.

Variables

var (
	// ErrNilSource indicates that a nil source was registered.
	ErrNilSource = errors.New("qi p2p: nil source")
	// ErrEmptyDocumentID indicates that a source has no stable document ID.
	ErrEmptyDocumentID = errors.New("qi p2p: empty document id")
	// ErrSourceAlreadyRegistered indicates that a document ID already has a source.
	ErrSourceAlreadyRegistered = errors.New("qi p2p: source already registered")
)

Functions

func Pull

func Pull(ctx context.Context, h host.Host, remote peer.AddrInfo, localReplicaID yin.ReplicaID, sink Sink) (bool, error)

Pull requests and applies generation-aware catch-up. localReplicaID is sent as caller-declared causal metadata, not as an authorization credential. Authorization at the source must use the authenticated libp2p peer supplied to SourceResolver; no caller-declared remote replica identity is required.

func ResolverStreamHandler

func ResolverStreamHandler(resolver SourceResolver) network.StreamHandler

ResolverStreamHandler returns the shared resolver-backed stream handler for the qi sync protocol. The transport bounds and decodes each frame, validates basic routing inputs, and then passes the authenticated remote peer and exact document ID to resolver. Store remains responsible for artifact identity, generation, yin ingest, and durable acceptance.

func Serve

func Serve(h host.Host, source Source) error

Serve registers the qi sync stream handler on h for one document. It is the single-document compatibility wrapper around ServeResolver; requests for any other document are unavailable rather than falling back to source.

func ServeResolver

func ServeResolver(h host.Host, resolver SourceResolver) error

ServeResolver registers one shared, resolver-backed qi sync stream handler on h. Resolver policy determines which remote peers may access each document; unavailable resolutions receive a single artifact-free response.

func StreamHandler

func StreamHandler(source Source) network.StreamHandler

StreamHandler returns a single-document libp2p stream handler. Requests for any other document are treated exactly like unavailable routed documents.

Types

type Sink

type Sink interface {
	Cursor() qi.Cursor
	ApplyCatchUp(qi.CatchUp) (bool, error)
}

Sink reports its generation-scoped cursor and applies a validated catch-up.

type Source

type Source interface {
	Identity() qi.DocumentIdentity
	CatchUp(qi.Cursor) (qi.CatchUp, error)
}

Source identifies exactly one qi document and chooses a same-generation delta or forced snapshot bootstrap. Identity().DocumentID is the sole route key and must remain stable while the Source is registered. CatchUp retains responsibility for validating document and generation identity and choosing snapshot versus delta. A Source served by a stream handler must permit concurrent Identity and CatchUp calls; Store provides that synchronization.

type SourceResolver

type SourceResolver interface {
	ResolveSource(remote peer.ID, documentID string) (Source, bool)
}

SourceResolver selects the source for an exact document ID in the context of the authenticated remote libp2p peer on the stream. The embedding application owns membership, peer/document authorization, and availability policy. ResolveSource may be called concurrently. It must return available false for unknown, removed, or denied documents and must never substitute another document; the transport exposes all unavailable outcomes identically.

type SourceResolverFunc

type SourceResolverFunc func(remote peer.ID, documentID string) (Source, bool)

SourceResolverFunc adapts a function to SourceResolver. The remote argument is the stream connection's authenticated libp2p peer, not a yin replica ID. The function may be invoked concurrently by stream handlers.

func (SourceResolverFunc) ResolveSource

func (f SourceResolverFunc) ResolveSource(remote peer.ID, documentID string) (Source, bool)

ResolveSource calls f with the authenticated remote peer and requested document ID.

type SourceRouter

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

SourceRouter is a concurrency-safe registry of one Source per document ID. Its zero value is ready for use. Unregister affects future resolutions only; the embedding application owns source lifetime, including sources returned by resolutions already in progress.

func NewSourceRouter

func NewSourceRouter() *SourceRouter

NewSourceRouter returns an empty source router.

func (*SourceRouter) Register

func (r *SourceRouter) Register(source Source) error

Register adds source under its stable document ID. Identity is read before taking the router lock, and its generation is intentionally ignored. The source must continue to report that document ID while it is registered.

func (*SourceRouter) ResolveSource

func (r *SourceRouter) ResolveSource(_ peer.ID, documentID string) (Source, bool)

ResolveSource returns the currently registered source for documentID. The remote peer is accepted to satisfy SourceResolver; SourceRouter itself does not apply peer authorization policy. Applications that need such policy should wrap it in another resolver. No router lock is held after return or while any Source method is called.

func (*SourceRouter) Unregister

func (r *SourceRouter) Unregister(documentID string) bool

Unregister removes the source for documentID from future resolutions and reports whether a source was removed. A source already returned by ResolveSource may still be in use: Unregister neither waits for in-flight calls nor manages the source's lifetime.

type Store

type Store interface {
	Source
	Sink
}

Store is the interface implemented by a one-document qi store.

type SyncRequest

type SyncRequest struct {
	RequesterReplicaID yin.ReplicaID `json:"requester_replica_id"`
	Cursor             qi.Cursor     `json:"cursor"`
}

SyncRequest is the request frame sent over the generation-aware protocol. RequesterReplicaID is caller-declared causal metadata, not an authenticated network identity or an authorization credential.

func (SyncRequest) MarshalJSON

func (r SyncRequest) MarshalJSON() ([]byte, error)

MarshalJSON preserves explicit yin version-vector counters.

func (*SyncRequest) UnmarshalJSON

func (r *SyncRequest) UnmarshalJSON(data []byte) error

UnmarshalJSON restores explicit yin version-vector counters.

type SyncResponse

type SyncResponse struct {
	CatchUp qi.CatchUp `json:"catch_up"`
	Error   string     `json:"error,omitempty"`
}

SyncResponse contains either a snapshot bootstrap, a delta, or no change.