perf: reduce allocations on hot paths (#38)

- Cache Prometheus WithLabelValues at init (metrics.go): eliminates
  per-height map lookups and slice allocs for sync state, backfill
  stages, and store operations.
- Move transient error needles to package-level var (celestia_node.go):
  avoids per-retry []string heap allocation.
- Cache bearer auth metadata map at construction (celestia_app.go):
  avoids map+string alloc per gRPC call.
- Replace map[string]any with struct in MarshalBlob (service.go):
  ~200 bytes less per blob marshal.
- Preallocate blob slices in GetBlobs, BlobGetAll, gRPC GetAll:
  reduces append doublings on typical result sets.
- Optimize notifier filterEvent: skip allocation when all/no blobs
  match; use exact-capacity slice for partial matches.
- Skip namespace map allocation for header-only subscriptions.
- Use slice indexing in httpToWS instead of TrimPrefix.

Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>

Author: tac0turtle

pkg/api/grpc/blob_service.go

@@ -73,7 +73,7 @@ func (s *BlobServiceServer) GetAll(ctx context.Context, req *pb.GetAllRequest) (
 		nsList[i] = ns
 	}
 
-	var allBlobs []types.Blob
+	allBlobs := make([]types.Blob, 0, len(nsList)*8)
 	for _, ns := range nsList {
 		blobs, err := s.svc.Store().GetBlobs(ctx, ns, req.Height, req.Height, 0, 0)
 		if err != nil {

pkg/api/notifier.go

@@ -77,9 +77,13 @@ func (n *Notifier) Subscribe(namespaces []types.Namespace) (*Subscription, error
 	}
 
 	id := n.nextID.Add(1)
-	nsSet := make(map[types.Namespace]struct{}, len(namespaces))
-	for _, ns := range namespaces {
-		nsSet[ns] = struct{}{}
+
+	var nsSet map[types.Namespace]struct{}
+	if len(namespaces) > 0 {
+		nsSet = make(map[types.Namespace]struct{}, len(namespaces))
+		for _, ns := range namespaces {
+			nsSet[ns] = struct{}{}
+		}
 	}
 
 	sub := &Subscription{
@@ -161,13 +165,27 @@ func (n *Notifier) Publish(event HeightEvent) {
 
 // filterEvent returns an event with blobs filtered to the subscriber's
 // namespace set. If the subscriber watches all namespaces, the event is
-// returned as-is.
+// returned as-is. Avoids allocation when no blobs match.
 func (n *Notifier) filterEvent(event HeightEvent, sub *Subscription) HeightEvent {
 	if len(sub.namespaces) == 0 {
 		return event
 	}
 
-	filtered := make([]types.Blob, 0, len(event.Blobs))
+	// Count matches first to avoid allocating when nothing matches.
+	count := 0
+	for i := range event.Blobs {
+		if _, ok := sub.namespaces[event.Blobs[i].Namespace]; ok {
+			count++
+		}
+	}
+	if count == len(event.Blobs) {
+		return event // all match — no copy needed
+	}
+	if count == 0 {
+		return HeightEvent{Height: event.Height, Header: event.Header}
+	}
+
+	filtered := make([]types.Blob, 0, count)
 	for i := range event.Blobs {
 		if _, ok := sub.namespaces[event.Blobs[i].Namespace]; ok {
 			filtered = append(filtered, event.Blobs[i])

pkg/api/service.go

@@ -70,7 +70,7 @@ func (s *Service) BlobGetByCommitment(ctx context.Context, commitment []byte) (j
 // limit=0 means no limit; offset=0 means no offset.
 // Pagination is applied to the aggregate result across all namespaces.
 func (s *Service) BlobGetAll(ctx context.Context, height uint64, namespaces []types.Namespace, limit, offset int) (json.RawMessage, error) {
-	var allBlobs []types.Blob
+	allBlobs := make([]types.Blob, 0, len(namespaces)*8) // preallocate for typical workload
 	for _, ns := range namespaces {
 		blobs, err := s.store.GetBlobs(ctx, ns, height, height, 0, 0)
 		if err != nil {
@@ -179,15 +179,25 @@ func (s *Service) Fetcher() fetch.DataFetcher {
 	return s.fetcher
 }
 
+// blobJSON is a struct-based representation for celestia-node compatible JSON.
+// Using a struct avoids the per-call map[string]any allocation that json.Marshal
+// requires for maps.
+type blobJSON struct {
+	Namespace    []byte `json:"namespace"`
+	Data         []byte `json:"data"`
+	ShareVersion uint32 `json:"share_version"`
+	Commitment   []byte `json:"commitment"`
+	Index        int    `json:"index"`
+}
+
 // MarshalBlob converts a stored blob into celestia-node compatible JSON.
 func MarshalBlob(b *types.Blob) json.RawMessage {
-	m := map[string]any{
-		"namespace":     b.Namespace[:],
-		"data":          b.Data,
-		"share_version": b.ShareVersion,
-		"commitment":    b.Commitment,
-		"index":         b.Index,
-	}
-	raw, _ := json.Marshal(m) //nolint:errcheck
+	raw, _ := json.Marshal(blobJSON{ //nolint:errcheck
+		Namespace:    b.Namespace[:],
+		Data:         b.Data,
+		ShareVersion: b.ShareVersion,
+		Commitment:   b.Commitment,
+		Index:        b.Index,
+	})
 	return raw
 }

pkg/fetch/celestia_app.go

@@ -29,11 +29,15 @@ type CelestiaAppFetcher struct {
 
 // bearerCreds implements grpc.PerRPCCredentials for bearer token auth.
 type bearerCreds struct {
-	token string
+	metadata map[string]string // cached; avoids allocation per RPC
+}
+
+func newBearerCreds(token string) bearerCreds {
+	return bearerCreds{metadata: map[string]string{"authorization": "Bearer " + token}}
 }
 
 func (b bearerCreds) GetRequestMetadata(_ context.Context, _ ...string) (map[string]string, error) {
-	return map[string]string{"authorization": "Bearer " + b.token}, nil
+	return b.metadata, nil
 }
 
 func (b bearerCreds) RequireTransportSecurity() bool { return false }
@@ -45,7 +49,7 @@ func NewCelestiaAppFetcher(grpcAddr, authToken string, log zerolog.Logger) (*Cel
 		grpc.WithTransportCredentials(insecure.NewCredentials()),
 	}
 	if authToken != "" {
-		opts = append(opts, grpc.WithPerRPCCredentials(bearerCreds{token: authToken}))
+		opts = append(opts, grpc.WithPerRPCCredentials(newBearerCreds(authToken)))
 	}
 
 	conn, err := grpc.NewClient(grpcAddr, opts...)

pkg/fetch/celestia_node.go

@@ -91,9 +91,9 @@ func NewCelestiaNodeFetcher(ctx context.Context, addr, token string, log zerolog
 func httpToWS(addr string) string {
 	switch {
 	case strings.HasPrefix(addr, "http://"):
-		return "ws://" + strings.TrimPrefix(addr, "http://")
+		return "ws://" + addr[len("http://"):]
 	case strings.HasPrefix(addr, "https://"):
-		return "wss://" + strings.TrimPrefix(addr, "https://")
+		return "wss://" + addr[len("https://"):]
 	default:
 		return addr
 	}
@@ -513,25 +513,28 @@ func isTransientRPCError(err error) bool {
 		return true
 	}
 	msg := strings.ToLower(err.Error())
-	for _, needle := range []string{
-		"eof",
-		"connection reset by peer",
-		"broken pipe",
-		"i/o timeout",
-		"timeout",
-		"temporarily unavailable",
-		"connection refused",
-		"503",
-		"504",
-		"502",
-	} {
+	for _, needle := range transientNeedles {
 		if strings.Contains(msg, needle) {
 			return true
 		}
 	}
 	return false
 }
 
+// transientNeedles is allocated once at package init to avoid per-call slice allocation.
+var transientNeedles = []string{
+	"eof",
+	"connection reset by peer",
+	"broken pipe",
+	"i/o timeout",
+	"timeout",
+	"temporarily unavailable",
+	"connection refused",
+	"503",
+	"504",
+	"502",
+}
+
 // jsonInt64 handles CometBFT's int64 fields encoded as JSON strings.
 type jsonInt64 int64