Built on go-p2p-forge

This page shows how go-ricochet is assembled from go-p2p-forge v0.2.1, so you can read the server’s source and add to it in a fork. For the framework itself, read forge’s user guide.

Forge is a Netty-style toolkit for libp2p servers. A pipeline is a list of middleware, each a func(sc *forge.StreamContext, next func()). Code before next() runs as the request goes in, code after it runs as the response comes out, and a middleware that sets sc.Err and returns without calling next() stops the request. go-ricochet builds one pipeline per protocol.

The server

internal/server/server.go builds a forge server with the UDX transport and starts it:

s.forgeServer = forge.NewServer(
  forge.WithConfig(forgeCfg),
  forge.WithIdentity(priv),
  forge.WithLogger(s.logger),
  forge.WithTransport(libp2p.Transport(udxtransport.NewTransport)),
)

buildForgeConfig() maps the server’s config onto forge.Config: port, listen and external addresses, bootstrap peers, Yamux timing, connection caps, relay and NAT settings, node.DHTModeServer and EnablePubSub. Starting the forge server creates the libp2p host, the Kademlia DHT and GossipSub.

Start then builds the server’s own components, puts the shared ones in forge’s registry, and registers one pipeline per protocol.

One agent, step by step

Here is the document agent (SDA) from internal/protocol/sda/handler.go. The other agents follow the same shape.

func NewPipeline(logger *slog.Logger, pool *codec.BufferPool, reg *forge.Registry) *forge.Pipeline {
  limiter := ratelimit.FromRegistry(reg).SDA

  routes := map[string]forge.Middleware{
    OpGET: middleware.Chain(forge.JSONDeserialize[DocRequest](), handleGet),
    OpPUT: middleware.Chain(forge.JSONDeserialize[DocRequest](), handlePut),
    // ... PATCH, HEAD, DELETE, LIST, BATCH_PUT, HISTORY, DIRECTORY, ACCESS
  }

  return wire.Bounded(forge.NewPipeline(logger,
    metrics.Middleware(metrics.FromRegistry(reg), "sda", ""),
    wire.AccessLog("sda", ""),
    middleware.Recovery(),
    docResponseWriter(),
    forge.FrameDecodeMiddleware(pool),
    wire.RequestDeadline(reg),
    middleware.DualRateLimitMiddleware(limiter, isWriteClassifier),
    admission.DrainGate(admission.DrainFromRegistry(reg)),
    admission.Middleware(admission.FromRegistry(reg)),
    capacity.WriteGate(capacity.FromRegistry(reg), isWriteClassifier),
    commonValidation(),
    middleware.OperationRouter("operation", routes),
  ).WithRegistry(reg), reg)
}

In order, from the outside in:

#	Middleware	Package	What it does
1	metrics.Middleware	go-ricochet internal/metrics	Outermost, so it counts panics too. Records ricochet_requests_total and ricochet_request_duration_seconds by protocol, operation and outcome.
2	wire.AccessLog	go-ricochet internal/protocol/wire	Writes the one Info request line per request after the response: protocol, op, status, peer, stream, duration.
3	middleware.Recovery	forge	Turns a panic in anything below into sc.Err.
4	docResponseWriter	the agent itself	After next() returns, turns sc.Err into a DocResponse with the status from wire.Classify and a safe message from wire.ClientMessage, then writes sc.Response as one JSON frame.
5	forge.FrameDecodeMiddleware	forge	Reads one length-prefixed frame into sc.RawBytes, using pooled buffers.
6	wire.RequestDeadline	go-ricochet	Puts the message_timeout_sec deadline on sc.Ctx, from here to the response.
7	middleware.DualRateLimitMiddleware	forge	Per-peer token buckets, one for reads and one for writes. isWriteClassifier peeks at the raw JSON to pick the bucket. The limiters are off unless configured.
8	admission.DrainGate	go-ricochet internal/admission	Refuses new requests with a 503 once shutdown has begun.
9	admission.Middleware	go-ricochet	Takes an admission slot (global and per peer) and releases it when the handler returns.
10	capacity.WriteGate	go-ricochet internal/capacity	Refuses writes with a 507 while the database is over its storage budget.
11	commonValidation	the agent itself	Peeks at the raw JSON, checks ownerPeerId and the path, stores the owner with sc.Set("ownerID", ...), and calls wire.RequireOwner for writes.
12	middleware.OperationRouter	forge	Dispatches on the operation field.

WithRegistry(reg) gives every StreamContext the server’s registry. wire.Bounded reads the server config from the registry and sets the pipeline’s idle timeout to connection_timeout_sec; forge’s own request timeout (forge.DefaultRequestTimeout, five minutes) stays as the ceiling on a stream’s life.

Authorization is not a separate middleware. The agents call helpers in internal/protocol/wire where they have the facts: RequireOwner and RequireSelf compare a claimed identity with sc.PeerID, the peer the connection authenticated, and RequireReader checks a document’s, feed’s or collection’s visibility and reader list. Each returns an error that wire.Classify maps to 403.

Operation routing

OperationRouter("operation", routes) parses the frame as a JSON object, reads the named field and calls the matching route. A missing field sets middleware.ErrMissingOperation and an unknown one middleware.ErrUnknownOperation; wire.Classify turns both into a 400. After a match it records the operation with sc.SetOperation, which is where the metrics and access-log middleware get their op label, so the label set is bounded by the routing table.

Each route is a middleware.Chain: here forge.JSONDeserialize[DocRequest]() unmarshals sc.RawBytes into sc.Request, then the handler runs. Chain stops at the first step that sets sc.Err. The mail agents MAA and MMA route on operationType instead; MSA serves one operation and has no router, so it passes a fallback operation name to metrics.Middleware and wire.AccessLog.

Handlers and the registry

A handler reads sc.Request, does its work through sc.Ctx, and sets sc.Response or sc.Err:

func handleHead(sc *forge.StreamContext, next func()) {
  req := sc.Request.(*DocRequest)
  store, _ := forge.ServiceFrom[storage.Storage](sc, "storage")
  ownerID, _ := sc.Get("ownerID")
  // ... store.HeadDocument(ctx, ownerID.(peer.ID), req.Path), then wire.RequireReader
}

The server registers its shared components once, in Start, before it builds the pipelines:

Registry key	Value	Read by
"storage"	storage.Storage (PostgreSQL)	SDA, SFA, SCA handlers
"mda"	*mda.MailboxServer	MAA, MMA handlers
"mta"	*mta.Router	MSA handlers
"config"	*core.ServerConfig	wire.Bounded, wire.RequestDeadline
ratelimit.RegistryKey ("ratelimits")	*ratelimit.Limiters	ratelimit.FromRegistry
admission.RegistryKey ("admission")	*admission.Controller	admission.FromRegistry
admission.DrainKey ("drain")	*admission.Drain	admission.DrainFromRegistry
metrics.RegistryKey ("metrics")	*metrics.Metrics	metrics.FromRegistry
capacity.RegistryKey ("capacity")	*capacity.Sampler	capacity.FromRegistry
trust.RegistryKey ("trust")	*trust.Peers	Nothing reads it from the registry in 1.1.0; the MTA router is given the set directly.

The FromRegistry helpers decide what a missing entry means. A missing admission controller, drain switch or capacity sampler admits everything, and missing metrics record nothing. Missing limiters fall back to the built-in defaults rather than to no limiter.

Registration

registerProtocolHandlers() builds each pipeline and attaches it to the libp2p host:

sdaPipeline := sda.NewPipeline(s.logger, pool, reg).WithActiveStreams(&s.inFlight)
h.SetStreamHandler(sda.ProtocolID, sdaPipeline.StreamHandler())

It calls SetStreamHandler on the host instead of forge’s Server.Handle, because Handle only records handlers for Start to install, and the server has to start forge first: the services need the host and node before the handlers are registered. WithActiveStreams(&s.inFlight) counts streams in progress, so Stop can wait for them before it closes storage.

Background work

The server’s background parts are its own types, started and stopped by hand in startServices and teardown, not forge Services:

• registry.Registry announces the server on /sf-network/services/announce and /sf-network/services/heartbeat through forge’s node.Node (JoinTopic, Publish).
• presence.Service publishes presence events on a per-server topic.
• mda.Notifier pushes notifications to recipients over notify.ProtocolID.
• relay.Relay relays the pubsub.relay_topics list on Node().PubSub().
• maintenanceLoop runs mailbox maintenance and the capacity sample on a ticker.

Forge concepts and where go-ricochet uses them

Forge concept	In go-ricochet 1.1.0
forge.NewServer, WithConfig, WithIdentity, WithTransport	internal/server/server.go initializeP2P
Identity helpers (host.LoadIdentityFromSeed, LoadIdentityFromFile, LoadOrCreateIdentity)	internal/server/server.go loadIdentity
forge.NewPipeline, StreamContext, middleware	Every agent: internal/protocol/ msa, maa, mma, sda, sfa, sca
FrameDecodeMiddleware, codec.BufferPool	Every agent; one pool shared by all, published as metrics by internal/metrics/collectors.go
JSONDeserialize[T]	Route chains in SDA; whole-pipeline in SFA and SCA
middleware.OperationRouter, middleware.Chain	MAA, MMA, SDA, SFA, SCA
middleware.Recovery	Every agent
RateLimitMiddleware, DualRateLimitMiddleware, TokenBucket, DualTokenBucket	internal/ratelimit/limiters.go and every agent
Registry, Provide, ServiceFrom, Service	internal/server/server.go Start; handlers; the FromRegistry helpers
Pipeline.WithIdleTimeout	internal/protocol/wire/timeouts.go Bounded
Pipeline.WithActiveStreams	registerProtocolHandlers
node.Node DHT and GossipSub	internal/registry/registry.go, internal/presence/service.go, internal/relay
forgetest.NewMockStreamWithFrame	internal/protocol/protocoltest/harness.go

Forge features go-ricochet 1.1.0 does not use: Server.Handle (see Registration), AddService and TickerService, OnPeerConnected (presence registers a libp2p notifiee directly), OpenStream (the notifier calls host.NewStream), MetricsCollector (the server keeps its own Prometheus registry), service.HealthCheck (the operator HTTP surface is internal/opsapi), and JSONResponseWriter (each agent has its own response writer so errors come back in its own envelope).