---
name: go-p2p-forge
description: 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.
triggers: go-p2p-forge, forge.NewServer, forge.NewPipeline, StreamContext, FrameDecodeMiddleware, JSONResponseWriter, JSONDeserialize, OperationRouter, forgetest, NewMockStreamWithFrame, TickerService, NewSingleBucket
---

# go-p2p-forge

## 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

```bash
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.

```go
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

### Minimal server

```go
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

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:

```go
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

```go
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

```go
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

```go
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

```go
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

```go
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

- `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

- **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

```go
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

- 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

- Overview: https://cryptopeer.net/ricochet/forge/
- User guide: https://cryptopeer.net/ricochet/forge/user-guide/
- Architecture: https://cryptopeer.net/ricochet/forge/architecture/
- Repository: https://github.com/stephanfeb/go-p2p-forge (`docs/USER_GUIDE.md`, `docs/architecture.md`)
- A large real user: go-ricochet `internal/protocol/*` and
  `internal/protocol/wire` (status mapping, `Classify`)
- Related skills: `ricochet-server`, `udx-go`