dmsgasync

package
v1.3.106 Latest Latest
Warning

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

Go to latest
Published: Nov 4, 2026 License: AGPL-3.0 Imports: 23 Imported by: 0

Documentation

Overview

Package dmsgasync pkg/dmsg/dmsgasync/client.go c1-net-dmsg

Package dmsgasync pkg/dmsg/dmsgasync/discovery.go c1-net-dmsg

Package dmsgasync pkg/dmsg/dmsgasync/dmsgasync.go c1-net-dmsg

Store and forward for offline dmsg keys. A recipient chooses inbox hosts, signs a Record naming them and registers with each. A sender that cannot reach the recipient deposits a payload on one of its hosts, and the recipient fetches it when it comes back. The host applies the recipient's sender Policy to the sender key the dmsg stream authenticates. See docs/design/dmsg-async.md.

Package dmsgasync pkg/dmsg/dmsgasync/host.go c1-net-dmsg

Package dmsgasync pkg/dmsg/dmsgasync/record.go c1-net-dmsg

Package dmsgasync pkg/dmsg/dmsgasync/store.go c1-net-dmsg

Index

Constants

View Source
const (
	DefaultMaxMessage        = 64 << 10
	DefaultMaxPerRecipient   = 200
	DefaultMaxRecipientBytes = 8 << 20
	DefaultMaxPerSender      = 20
	DefaultMaxTotal          = 256 << 20
	DefaultTTL               = 7 * 24 * time.Hour
	// MaxMessageLimit caps any message size, since a fetch reply carries at
	// least one whole message.
	MaxMessageLimit = 1 << 20
	// MaxRecordHosts bounds the hosts one Record may name.
	MaxRecordHosts = 4
	// MaxRecordLifetime bounds how far ahead a Record may expire.
	MaxRecordLifetime = 30 * 24 * time.Hour
)

Defaults for a zero field in Limits.

View Source
const (
	OpRegister   = "register"
	OpUnregister = "unregister"
	OpRecord     = "record"
	OpDeposit    = "deposit"
	OpFetch      = "fetch"
	OpAck        = "ack"
)

Operations a client sends.

View Source
const (
	StoreDisk   = "disk"
	StoreMemory = "memory"
)

Store backends.

Port is the dmsg port inbox hosts serve.

Variables

View Source
var (
	ErrNotHosted     = errors.New("dmsg-async: host does not hold messages for this key")
	ErrNotAllowed    = errors.New("dmsg-async: the recipient does not accept messages from this sender")
	ErrTooBig        = errors.New("dmsg-async: message is larger than the host takes")
	ErrRecipientFull = errors.New("dmsg-async: the recipient has too much waiting on this host")
	ErrSenderFull    = errors.New("dmsg-async: this sender has too much waiting for the recipient")
	ErrStoreFull     = errors.New("dmsg-async: host store is full")
	ErrBadRecord     = errors.New("dmsg-async: invalid inbox record")
	ErrBadRequest    = errors.New("dmsg-async: malformed request")
	ErrInternal      = errors.New("dmsg-async: host error")
)

Errors an operation can end with. The host sends a code and the client maps it back to one of these.

View Source
var ErrNoRecord = errors.New("dmsg-async: no inbox record")

ErrNoRecord means dmsg discovery has no inbox record for the key.

Functions

func Ack

func Ack(ctx context.Context, dial Dialer, host cipher.PubKey, ids []string) (int, error)

Ack removes the dialer's messages that ids name from host and reports how many it removed.

func Deposit

func Deposit(ctx context.Context, dial Dialer, host, to cipher.PubKey, port uint16, ref string, payload []byte) (string, error)

Deposit leaves payload for `to` on port with host and returns its id.

func DepositAny

func DepositAny(ctx context.Context, dial Dialer, rec Record, port uint16, ref string, payload []byte) (cipher.PubKey, error)

DepositAny deposits on rec's hosts in order until one takes it, and returns that host. It fails with the last error when none does.

func Fetch

func Fetch(ctx context.Context, dial Dialer, host cipher.PubKey, ports []uint16, handle Handler) (int, error)

Fetch takes what waits for the dialer on host for ports, hands each message to handle and acks the ones it took. It repeats while the host has more and reports how many were handled.

func NewRef

func NewRef() string

NewRef returns a fresh deposit reference. A sender reuses one ref for every retry of the same message so a host stores it once.

func PublishRecord

func PublishRecord(ctx context.Context, hc *http.Client, discURL string, rec Record) error

PublishRecord posts rec to the dmsg discovery at discURL.

func Register

func Register(ctx context.Context, dial Dialer, host cipher.PubKey, rec Record, pol Policy) error

Register registers rec and pol with host. rec MUST be signed by the dialer's own key and name host.

func RemotePK

func RemotePK(conn net.Conn) (cipher.PubKey, bool)

RemotePK is the authenticated key at the other end of a dmsg stream.

func Unregister

func Unregister(ctx context.Context, dial Dialer, host cipher.PubKey) (int, error)

Unregister removes the dialer's registration from host and drops what waits for it there.

Types

type Dialer

type Dialer func(ctx context.Context, pk cipher.PubKey, port uint16) (net.Conn, error)

Dialer opens a dmsg stream to pk on port.

type Handler

type Handler func(Message) error

Handler takes one fetched message. A nil error acks it. An error leaves it on the host for a later fetch.

type Header struct {
	ID        string        `json:"id"`
	From      cipher.PubKey `json:"from"`
	Port      uint16        `json:"port"`
	Size      int           `json:"size"`
	Deposited time.Time     `json:"deposited"`
}

Header describes one stored message.

type Host

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

Host serves the dmsg-async protocol for the keys registered with it.

func NewHost

func NewHost(cfg HostConfig) (*Host, error)

NewHost opens the store and loads the registrations a disk host kept.

func (*Host) Handle

func (h *Host) Handle(conn net.Conn, remote cipher.PubKey)

Handle runs one operation from remote on conn and closes it. remote MUST be the key the stream authenticated.

func (*Host) Limits

func (h *Host) Limits() Limits

Limits returns the host's limits with defaults applied.

func (*Host) Record

func (h *Host) Record(pk cipher.PubKey) (Record, bool)

Record returns pk's registered record while it is unexpired.

func (*Host) Registered

func (h *Host) Registered() []cipher.PubKey

Registered lists the keys registered with this host.

func (*Host) Serve

func (h *Host) Serve(ctx context.Context, l net.Listener) error

Serve answers streams from l until it closes, and sweeps expired messages.

func (*Host) Stats

func (h *Host) Stats() Stats

Stats reports what the store holds.

type HostConfig

type HostConfig struct {
	// PK is the host's own key, which a record must name.
	PK cipher.PubKey
	// Store is StoreDisk (the default) or StoreMemory.
	Store string
	// Dir holds the disk store and the registrations. Required for StoreDisk.
	Dir    string
	Limits Limits
	// Accepts reports whether this host agrees to hold for a key. It is
	// asked when the key registers, and a key it refuses cannot register.
	Accepts func(cipher.PubKey) bool
	Log     *logging.Logger
}

HostConfig configures an inbox host.

type Limits

type Limits struct {
	MaxMessage        int           `json:"max_message,omitempty"`
	MaxPerRecipient   int           `json:"max_per_recipient,omitempty"`
	MaxRecipientBytes int64         `json:"max_recipient_bytes,omitempty"`
	MaxTotal          int64         `json:"max_total_size,omitempty"`
	TTL               time.Duration `json:"ttl,omitempty"`
}

Limits bounds a host's store. A zero field takes its default.

func (Limits) WithDefaults

func (l Limits) WithDefaults() Limits

WithDefaults returns l with each zero field at its default.

type Message

type Message struct {
	Header
	Payload []byte
	// Host is the inbox host that held it.
	Host cipher.PubKey
}

Message is a fetched message.

func List

func List(ctx context.Context, dial Dialer, host cipher.PubKey, ports []uint16) ([]Message, bool, error)

List returns one batch of what waits for the dialer on host for ports, oldest first, without removing it, and whether more wait. Ack removes what the caller took.

type Policy

type Policy struct {
	// Allow lists the senders that may deposit. Empty allows every key.
	Allow []cipher.PubKey `json:"allow,omitempty"`
	// Deny lists senders that may never deposit.
	Deny []cipher.PubKey `json:"deny,omitempty"`
	// MaxPerSender bounds what one sender may have waiting. Zero takes
	// DefaultMaxPerSender.
	MaxPerSender int `json:"max_per_sender,omitempty"`
	// PortAllow narrows who may deposit on the ports it names: only the listed
	// senders, and only when Allow and Deny let them.
	PortAllow map[uint16][]cipher.PubKey `json:"port_allow,omitempty"`
}

Policy is the recipient's rule for who may deposit. Only its hosts see it.

func (Policy) Allows

func (p Policy) Allows(sender cipher.PubKey) bool

Allows reports whether sender may deposit.

func (Policy) AllowsOn

func (p Policy) AllowsOn(sender cipher.PubKey, port uint16) bool

AllowsOn reports whether sender may deposit on port.

func (Policy) Equal

func (p Policy) Equal(q Policy) bool

Equal reports whether p and q allow the same senders with the same limit.

type Record

type Record struct {
	PK         cipher.PubKey   `json:"pk"`
	Hosts      []cipher.PubKey `json:"hosts"`
	MaxMessage int             `json:"max_message"`
	Seq        uint64          `json:"seq"`
	Expires    time.Time       `json:"expires"`
	Sig        cipher.Sig      `json:"sig"`
}

Record is a recipient's signed statement of where to deposit for it. It names hosts, not senders, so publishing it does not reveal its contacts.

func FetchRecord

func FetchRecord(ctx context.Context, hc *http.Client, discURL string, pk cipher.PubKey) (Record, error)

FetchRecord reads pk's inbox record from the dmsg discovery at discURL and verifies it.

func LookupRecord

func LookupRecord(ctx context.Context, dial Dialer, host, pk cipher.PubKey) (Record, error)

LookupRecord asks host for pk's current record and verifies it.

func NewRecord

func NewRecord(pk cipher.PubKey, sk cipher.SecKey, hosts []cipher.PubKey, maxMessage int, seq uint64, lifetime time.Duration) (Record, error)

NewRecord builds and signs a record for pk.

func (Record) Names

func (r Record) Names(host cipher.PubKey) bool

Names reports whether host is one of r's hosts.

func (*Record) Sign

func (r *Record) Sign(sk cipher.SecKey) error

Sign fills Sig with sk, which must be the key of r.PK.

func (Record) Verify

func (r Record) Verify(now time.Time) error

Verify checks the signature, the bounds and the expiry at now.

type Stats

type Stats struct {
	Messages   int   `json:"messages"`
	Bytes      int64 `json:"bytes"`
	Recipients int   `json:"recipients"`
}

Stats is what a store holds.

Jump to

Keyboard shortcuts

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