mirror of
https://github.com/luxfi/warp.git
synced 2026-10-30 15:13:04 +00:00
zeekay
251cd5dc55
Replaces github.com/luxfi/codec/wrappers.Errs with a tiny per-package
firstErr helper that preserves the historical "first error wins"
semantics:
- warp.firstErr (eventsocket.go, chainipc.go)
- socket.firstErr (socket/socket.go)
No package in this module used wrappers.Packer; only wrappers.Errs. The
new helper is identity-equal for the same input sequence so on-disk /
on-wire behavior is unchanged.
Adds firsterr_test.go in both packages with:
- referenceFirstErr mirroring the historical wrappers.Errs.Add
behavior
- cross-check across nil / single / multi / seeded cases including
the seeded socket.Close pattern
Drops github.com/luxfi/codec from the direct require in go.mod.
131 lines
3.2 KiB
Go
131 lines
3.2 KiB
Go
// Copyright (C) 2019-2025, Lux Industries, Inc. All rights reserved.
|
|
// See the file LICENSE for licensing terms.
|
|
|
|
package warp
|
|
|
|
import (
|
|
"fmt"
|
|
"maps"
|
|
"path/filepath"
|
|
"slices"
|
|
|
|
"github.com/luxfi/ids"
|
|
log "github.com/luxfi/log"
|
|
)
|
|
|
|
const (
|
|
// DefaultBaseURL can be used as a reasonable default value for the base URL
|
|
DefaultBaseURL = "/tmp"
|
|
|
|
ipcIdentifierPrefix = "ipc"
|
|
ipcConsensusIdentifier = "consensus"
|
|
ipcDecisionsIdentifier = "decisions"
|
|
)
|
|
|
|
type ipcContext struct {
|
|
log log.Logger
|
|
networkID uint32
|
|
path string
|
|
}
|
|
|
|
// ChainIPCs maintains IPCs for a set of chains
|
|
type ChainIPCs struct {
|
|
ipcContext
|
|
chains map[ids.ID]*EventSockets
|
|
blockAcceptorGroup AcceptorGroup
|
|
txAcceptorGroup AcceptorGroup
|
|
vertexAcceptorGroup AcceptorGroup
|
|
}
|
|
|
|
// NewChainIPCs creates a new *ChainIPCs that writes consensus and decision
|
|
// events to IPC sockets
|
|
func NewChainIPCs(
|
|
log log.Logger,
|
|
path string,
|
|
networkID uint32,
|
|
blockAcceptorGroup AcceptorGroup,
|
|
txAcceptorGroup AcceptorGroup,
|
|
vertexAcceptorGroup AcceptorGroup,
|
|
defaultChainIDs []ids.ID,
|
|
) (*ChainIPCs, error) {
|
|
cipcs := &ChainIPCs{
|
|
ipcContext: ipcContext{
|
|
log: log,
|
|
networkID: networkID,
|
|
path: path,
|
|
},
|
|
chains: make(map[ids.ID]*EventSockets),
|
|
blockAcceptorGroup: blockAcceptorGroup,
|
|
txAcceptorGroup: txAcceptorGroup,
|
|
vertexAcceptorGroup: vertexAcceptorGroup,
|
|
}
|
|
for _, chainID := range defaultChainIDs {
|
|
if _, err := cipcs.Publish(chainID); err != nil {
|
|
return nil, err
|
|
}
|
|
}
|
|
return cipcs, nil
|
|
}
|
|
|
|
// Publish creates a set of eventSockets for the given chainID
|
|
func (cipcs *ChainIPCs) Publish(chainID ids.ID) (*EventSockets, error) {
|
|
if es, ok := cipcs.chains[chainID]; ok {
|
|
cipcs.log.Info("returning existing event sockets",
|
|
log.Stringer("blockchainID", chainID),
|
|
)
|
|
return es, nil
|
|
}
|
|
|
|
es, err := newEventSockets(
|
|
cipcs.ipcContext,
|
|
chainID,
|
|
cipcs.blockAcceptorGroup,
|
|
cipcs.txAcceptorGroup,
|
|
cipcs.vertexAcceptorGroup,
|
|
)
|
|
if err != nil {
|
|
cipcs.log.Error("can't create ipcs",
|
|
log.Reflect("error", err),
|
|
)
|
|
return nil, err
|
|
}
|
|
|
|
cipcs.chains[chainID] = es
|
|
cipcs.log.Info("created IPC sockets",
|
|
log.Stringer("blockchainID", chainID),
|
|
log.UserString("consensusURL", es.ConsensusURL()),
|
|
log.UserString("decisionsURL", es.DecisionsURL()),
|
|
)
|
|
return es, nil
|
|
}
|
|
|
|
// Unpublish stops the eventSocket for the given chain if it exists. It returns
|
|
// whether or not the socket existed and errors when trying to close it
|
|
func (cipcs *ChainIPCs) Unpublish(chainID ids.ID) (bool, error) {
|
|
chainIPCs, ok := cipcs.chains[chainID]
|
|
if !ok {
|
|
return false, nil
|
|
}
|
|
delete(cipcs.chains, chainID)
|
|
return true, chainIPCs.stop()
|
|
}
|
|
|
|
// GetPublishedBlockchains returns the chains that are currently being published
|
|
func (cipcs *ChainIPCs) GetPublishedBlockchains() []ids.ID {
|
|
return slices.Collect(maps.Keys(cipcs.chains))
|
|
}
|
|
|
|
func (cipcs *ChainIPCs) Shutdown() error {
|
|
cipcs.log.Info("shutting down chain IPCs")
|
|
|
|
errs := firstErr{}
|
|
for _, ch := range cipcs.chains {
|
|
errs.add(ch.stop())
|
|
}
|
|
return errs.err
|
|
}
|
|
|
|
func ipcURL(ctx ipcContext, chainID ids.ID, eventType string) string {
|
|
return filepath.Join(ctx.path, fmt.Sprintf("%d-%s-%s", ctx.networkID, chainID.String(), eventType))
|
|
}
|