Skip to content
cryptopeercryptopeer.net

go-p2p-forge

Skill go-p2p-forge. Use when building a Go libp2p service with go-p2p-forge (module github.com/stephanfeb/go-p2p-forge): forge.NewServer, middleware Pipeline and StreamContext, framed JSON codecs and buffer pools, OperationRouter, rate limiters, lifecycle services and health checks, the host and node packages, and testing handlers with forgetest.

SKILL.md (Claude Code, Codex, Cursor) · go-p2p-forge.md (Boubou) · install: curl -fsSL https://cryptopeer.net/skills/install.sh | sh -s -- go-p2p-forge

What it is

Section titled “What it is”

go-p2p-forge is a Netty-style framework for libp2p servers in Go: a middleware pipeline per protocol (code before next() is the request path, code after it the response path), a length-prefixed frame codec with pooled buffers, a server builder that wires identity, host, DHT and GossipSub, an ordered service lifecycle, rate limiters and test mocks. go-ricochet’s protocol handlers are built on it.

Use plain go-libp2p for a client or a one-off stream handler; forge pays off for a long-running server with several request/response protocols. For UDX transport details see udx-go.

Install

Section titled “Install”
Terminal window
go get github.com/stephanfeb/go-p2p-forge@v0.2.2

go.mod declares Go 1.25.7; it builds against go-libp2p v0.49.0, go-libp2p-kad-dht v0.42.2 and go-libp2p-pubsub v0.15.0.

import (
    forge "github.com/stephanfeb/go-p2p-forge"
    "github.com/stephanfeb/go-p2p-forge/codec"
    "github.com/stephanfeb/go-p2p-forge/forgetest" // tests only
    forgehost "github.com/stephanfeb/go-p2p-forge/host"
    "github.com/stephanfeb/go-p2p-forge/middleware"
    "github.com/stephanfeb/go-p2p-forge/node"
    "github.com/stephanfeb/go-p2p-forge/service"
)

Core workflow

Section titled “Core workflow”

Minimal server

Section titled “Minimal server”
type PingRequest struct{ Message string `json:"message"` }
type PingResponse struct{ Reply string `json:"reply"` }


func main() {
    logger := slog.New(slog.NewTextHandler(os.Stdout, nil))
    pool := codec.NewBufferPool()
    limiter := middleware.NewSingleBucket(time.Minute, 100)
    defer limiter.Close()


    srv := forge.NewServer(
        forge.WithConfig(forge.DefaultConfig()),  // first: it replaces the whole config
        forge.WithLogger(logger),
        forge.WithTransport(libp2p.Transport(tcp.NewTCPTransport)),
        forge.WithListenAddrs("/ip4/0.0.0.0/tcp/9000"),
        forge.WithShutdownTimeout(10*time.Second),
    )


    pipeline := forge.NewPipeline(logger,
        middleware.Recovery(),
        forge.JSONResponseWriter(),         // writes sc.Response on the way back
        forge.FrameDecodeMiddleware(pool),  // reads one frame into sc.RawBytes
        middleware.RateLimitMiddleware(limiter),
        forge.JSONDeserialize[PingRequest](),
        func(sc *forge.StreamContext, next func()) {
            req := sc.Request.(*PingRequest)
            sc.Response = &PingResponse{Reply: "pong: " + req.Message}
        },
    ).WithActiveStreams(srv.ActiveStreams()).WithRegistry(srv.Registry())


    srv.Handle("/my-app/ping/1.0.0", pipeline.StreamHandler())


    ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
    defer cancel()
    if err := srv.ListenAndServe(ctx); err != nil { // Start, wait for ctx, Stop
        logger.Error("server error", "error", err)
        os.Exit(1)
    }
}

Besides the forge packages it imports context, log/slog, os, os/signal, syscall, time, github.com/libp2p/go-libp2p and github.com/libp2p/go-libp2p/p2p/transport/tcp.

For UDX pass forge.WithTransport(libp2p.Transport(udxtransport.NewTransport)) (udxtransport "github.com/stephanfeb/go-libp2p-udx-transport") and a /udp/<port>/udx listen address.

Turn pipeline errors into a response

Section titled “Turn pipeline errors into a response”

Every built-in failure (bad frame, rate limit, bad JSON, unknown operation, panic) only sets sc.Err; with no sc.Response the stream just closes. Put an error mapper directly below the response writer:

type ErrorResponse struct {
    Error        string `json:"error"`
    RetryAfterMs int64  `json:"retryAfterMs,omitempty"`
}


func errorResponse(sc *forge.StreamContext, next func()) {
    next()
    if sc.Err == nil {
        return
    }
    resp := &ErrorResponse{Error: "request failed"}
    var rl *forge.RateLimitedError
    if errors.As(sc.Err, &rl) {
        resp.Error, resp.RetryAfterMs = "rate limited", rl.RetryAfter.Milliseconds()
    } else if errors.Is(sc.Err, forge.ErrRateLimited) {
        resp.Error = "rate limited"
    }
    sc.Response = resp // JSONResponseWriter, above us, writes it
}


forge.NewPipeline(logger, middleware.Recovery(), forge.JSONResponseWriter(),
    errorResponse, forge.FrameDecodeMiddleware(pool), /* ... */)

Do not echo sc.Err.Error() to peers; it can carry internal detail.

Several operations on one protocol

Section titled “Several operations on one protocol”
routes := map[string]forge.Middleware{
    "get": middleware.Chain(forge.JSONDeserialize[GetReq](), getHandler),
    "put": middleware.Chain(forge.JSONDeserialize[PutReq](), putHandler),
}
p := forge.NewPipeline(logger,
    middleware.Recovery(), forge.JSONResponseWriter(), errorResponse,
    forge.FrameDecodeMiddleware(pool),
    middleware.OperationRouter("op", routes), // {"op":"get",...}
)
// In a handler, sc.Operation() is the routed name (only set after a match).

Dependency injection

Section titled “Dependency injection”
srv.Provide("store", myStore) // before Start; a duplicate key panics
// pipeline.WithRegistry(srv.Registry()) is required, or lookups fail
func getHandler(sc *forge.StreamContext, next func()) {
    st, ok := forge.ServiceFrom[*Store](sc, "store")
    if !ok { sc.Err = errors.New("no store"); return }
    v, err := st.Get(sc.Ctx, sc.Request.(*GetReq).Key) // sc.Ctx is cancelled with the stream
    ...
}

Multi-frame responses

Section titled “Multi-frame responses”
sc.Response = forge.NewSliceIterator(forge.JSONCodec{}, header, item1, item2)
// or any type implementing forge.FrameIterator { Next() ([]byte, error) }, io.EOF ends it

Server push and pubsub

Section titled “Server push and pubsub”
sc, err := srv.OpenStream(ctx, peerID, "/my-app/notify/1.0.0") // after Start
if err == nil {
    defer sc.Stream.Close()
    _ = codec.WriteFrame(sc.Stream, payload)
}


_ = srv.Node().JoinTopic("/my-app/events")        // required before Publish/Subscribe
_ = srv.Node().Publish(ctx, "/my-app/events", data)
sub := srv.Node().Subscribe("/my-app/events")     // nil if not joined

Background work and health

Section titled “Background work and health”
ticker := service.NewTickerService("announce", 30*time.Second,
    func(ctx context.Context) error { return announce(ctx) }, logger)
srv.AddService(ticker)                           // started in order, stopped in reverse


hc := service.NewHealthCheck("127.0.0.1:8080", logger) // /healthz, /readyz
hc.AddCheck(func() error {
    if !ticker.Healthy() { return fmt.Errorf("announce: %v", ticker.LastError()) }
    return nil
})
srv.AddService(hc)

Connection events: forge.OnPeerConnected(func(peer.ID)) and forge.OnPeerDisconnected(...); each runs in its own goroutine with panic recovery.

Key concepts / API map

Section titled “Key concepts / API map”
  • forge.Server - NewServer(opts...), Handle, AddService, Provide, Registry, Start, Stop, ListenAndServe, Host, Node, PeerID, OpenStream, ActiveStreams, Metrics.
  • Options: WithConfig, WithLogger, WithIdentity, WithPort, WithListenAddrs, WithBootstrapPeers, WithTransport, WithPreset (DevelopmentPreset, ProductionPreset), WithMetrics, WithShutdownTimeout, OnPeerConnected, OnPeerDisconnected.
  • forge.Config - Host forgehost.Config, Node node.Config, IdentityFile, DataDirectory (./data), RateLimitWindow, MaxRequestsPerWindow, ShutdownTimeout (5 s); DefaultConfig(), Validate(), LoadConfigFromFile(path, base).
  • forge.Pipeline - NewPipeline, Use, WithRegistry, WithActiveStreams, WithIdleTimeout (30 s), WithRequestTimeout (5 min), HandleStream, StreamHandler.
  • forge.StreamContext - Ctx, Stream, PeerID, Logger, RawBytes, Request, Response, Err, PoolBuf, Registry, IdleTimeout; Set/Get, Operation/SetOperation, WriteFrame.
  • Codec: forge.Codec interface, JSONCodec, DeserializeMiddleware[T], ResponseWriterMiddleware, JSONDeserialize[T], JSONResponseWriter, FrameDecodeMiddleware, FrameIterator, NewSliceIterator.
  • codec - ReadFrame, WriteFrame, ReadFrameWithTimeout, WriteFrameWithTimeout, ReadFramePooled, NewBufferPool, PoolBuffer (Bytes, Release), MaxFrameSize (10 MB), LengthPrefixSize (4).
  • middleware - Recovery, RateLimitMiddleware, RateLimitCostMiddleware, DualRateLimitMiddleware, DualRateLimitCostMiddleware, MetricsMiddleware, OperationRouter, Chain; limiters NewSingleBucket(window, max), NewDualBucket(window, maxRead, maxWrite), NewTokenBucket(window, rate, burst) and NewDualTokenBucket(...) (these report RetryAfter); errors ErrMissingOperation, ErrUnknownOperation, ErrPanic.
  • Errors in forge: ErrPanic, ErrRateLimited, ErrServerNotStarted, *RateLimitedError{RetryAfter}.
  • service - Service interface (Name, Start(ctx), Stop), Lifecycle, NewTickerService, NewHealthCheck.
  • host (import as forgehost) - Config, DefaultConfig, Create, RelayLimits (incl. UnlimitedConnections), LoadOrCreateIdentity, LoadIdentityFromFile, LoadIdentityFromSeed, PeerIDFromIdentity, DefaultYamuxMaxIncomingStreams (512).
  • node - Node (DHT, PubSub, JoinTopic, LeaveTopic, Publish, Subscribe, LogDHTStatus, Close), Config, DHTModeServer|Client|Auto.
  • forgetest - MockStream, NewMockStream, NewMockStreamWithFrame, ReadResponseFrame, MockConn, PipeStream / NewPipeStream, GenerateTestPeerID.

Gotchas

Section titled “Gotchas”
  • Middleware order matters. Recovery first, then JSONResponseWriter (it acts after next() returns), then error mapping, then FrameDecodeMiddleware; anything that reads sc.RawBytes (router, cost limiter, dual limiter) must come after the frame decoder.
  • Failures send nothing by default. Rate limit, decode and router errors set sc.Err and skip the rest; a peer sees the stream close with no frame. Add an error-to-response middleware (above).
  • Set sc.Err to stop the chain. run checks sc.Err before every middleware, and Chain checks it between steps; the code after next() in upper middleware still runs.
  • A single-frame marshal or write failure is only logged, not put in sc.Err, so MetricsMiddleware records it as success.
  • middleware.ErrPanic and forge.ErrPanic are different values. Recovery() sets middleware.ErrPanic; check that one with errors.Is.
  • sc.RawBytes is a pooled buffer released when the stream ends. Copy it before handing it to a goroutine or keeping it.
  • WithConfig replaces the whole config, so it must come before WithPort, WithListenAddrs and friends. Build configs from forge.DefaultConfig(): a bare &forge.Config{...} has MaxRequestsPerWindow 0 and fails Validate(), and turns relay and AutoNAT off.
  • WithPort alone listens on /udp/<port>/udx, but without WithTransport the host has only go-libp2p’s default transports, which have no UDX. Pass the UDX transport, or use WithListenAddrs with a transport that matches. The README quick start skips this.
  • WithTransport disables the default transports entirely; list every transport you listen on.
  • WithBootstrapPeers sets Node.BootstrapPeers (DHT). Static relays for EnableAutoRelay come from Host.BootstrapPeers, set through the config.
  • Relay flags depend on EnableRelay. Auto-relay, hole punching and the relay service without it fail Validate(). The relay service forces public reachability, and relayed connections are cut at go-libp2p’s 2 min / 128 KiB unless RelayLimits.UnlimitedConnections is set (needed when peers talk through the relay rather than only hole punch).
  • Config rate-limit fields and presets do nothing on their own. The server never builds a limiter from RateLimitWindow / MaxRequestsPerWindow; create limiters yourself and Close() them on shutdown. WithMetrics only stores the collector: add middleware.MetricsMiddleware(srv.Metrics()).
  • Graceful drain needs both WithShutdownTimeout(d > 0) and pipeline.WithActiveStreams(srv.ActiveStreams()) on every pipeline. Stop() then removes handlers, waits, stops services in reverse, and closes the node (which closes the host).
  • Register handlers and Provide before Start. Handle after Start is not applied to the host. Host(), Node() and PeerID() are empty until then; OpenStream returns ErrServerNotStarted.
  • OpenStream streams have no idle timeout (IdleTimeout zero) and the caller closes them. Set a context deadline.
  • Use sc.WriteFrame, not codec.WriteFrame(sc.Stream, ...), in response code, so the idle timeout covers a peer that stops reading.
  • Frames are capped at 10 MB and empty frames are rejected. Large frames grow as bytes arrive, so a peer that declares 10 MB and sends nothing costs nothing.
  • Identity: with no WithIdentity, the key is loaded or created at DataDirectory/peer_identity.key (or IdentityFile). Keep it on persistent storage or the peer ID changes on every deploy.
  • Per-peer rate limits bound well-behaved peers only: peer IDs are free. MaxConnectionsPerIP is 0 by default on purpose (carrier NAT).
  • The DHT admits a peer to its routing table only over a connection this host opened; inbound-only clients never appear there.
  • The README says MIT; LICENSE is Apache 2.0.

Testing

Section titled “Testing”
func TestPing(t *testing.T) {
    p := forge.NewPipeline(slog.Default(),
        middleware.Recovery(), forge.JSONResponseWriter(),
        forge.FrameDecodeMiddleware(codec.NewBufferPool()),
        forge.JSONDeserialize[PingRequest](), pingHandler)


    req, _ := json.Marshal(PingRequest{Message: "hi"})
    s := forgetest.NewMockStreamWithFrame(forgetest.GenerateTestPeerID(), req)
    p.HandleStream(s)


    out, err := s.ReadResponseFrame()
    if err != nil { t.Fatal(err) }
    var resp PingResponse
    _ = json.Unmarshal(out, &resp)
    if resp.Reply != "pong: hi" { t.Fatalf("got %q", resp.Reply) }
}
  • MockStream ignores deadlines and returns EOF at the end of its buffer. For timeout behaviour use forgetest.NewPipeStream(peerID), which blocks and honours deadlines; drive the returned net.Conn from the test and run HandleStream in a goroutine.
  • Test a middleware alone by building a &forge.StreamContext{Stream: s, PeerID: pid, Logger: slog.Default()} and passing a next that records it was called.
  • With DI, build forge.NewRegistry(), Provide fakes and call pipeline.WithRegistry(reg).
  • End-to-end, start two servers on /ip4/127.0.0.1/tcp/0 with WithTransport(libp2p.Transport(tcp.NewTCPTransport)), WithIdentity and a t.TempDir() data directory, as forge_test.go does.
  • Always go test ./... -race -count=1.

Red lines

Section titled “Red lines”
  • Never trust identity from a request body; authorise on sc.PeerID, which the Noise or TLS handshake authenticated.
  • Never return raw internal errors (SQL text, file paths) to peers.
  • Never retain sc.RawBytes or sc.PoolBuf past the handler.
  • Never mutate yamux.DefaultTransport.Config() in place; host.Create copies it because every host in the process shares it.
  • Never leave a limiter, ticker or outbound stream unclosed.
  • Never change a deployed protocol ID or frame format; add a new version.

Reference

Section titled “Reference”

Built by Stephan Februarygithub.com/stephanfebLinkedIn