Repository: lightninglabs/taproot-assets
Pull request #338 (2023-06-01 21:53:36)

universe: fix potential loop scoping bugs, add additional debug logging 
In this commit we rename the `chanutils` package to `fn`, as it's much shorter, and how houses generic functions that deal with more than just channels.

We then attempt to fix some potential loop scoping bugs in the universe code. Finally, we add some logging to print out the id+key for the fail universe log sync events. 


Patch:
diff --git a/address/book.go b/address/book.go
index c9a20134..f5089355 100644
--- a/address/book.go
+++ b/address/book.go
@@ -11,8 +11,8 @@ import (
 	"github.com/btcsuite/btcd/wire"
 	"github.com/lightninglabs/lndclient"
 	"github.com/lightninglabs/taproot-assets/asset"
-	"github.com/lightninglabs/taproot-assets/chanutils"
 	"github.com/lightninglabs/taproot-assets/commitment"
+	"github.com/lightninglabs/taproot-assets/fn"
 	"github.com/lightningnetwork/lnd/keychain"
 )
 
@@ -150,7 +150,7 @@ type Book struct {
 
 	// subscribers is a map of components that want to be notified on new
 	// address events, keyed by their subscription ID.
-	subscribers map[uint64]*chanutils.EventReceiver[*AddrWithKeyInfo]
+	subscribers map[uint64]*fn.EventReceiver[*AddrWithKeyInfo]
 
 	// subscriberMtx guards the subscribers map and access to the
 	// subscriptionID.
@@ -158,15 +158,15 @@ type Book struct {
 }
 
 // A compile-time assertion to make sure Book satisfies the
-// chanutils.EventPublisher interface.
-var _ chanutils.EventPublisher[*AddrWithKeyInfo, QueryParams] = (*Book)(nil)
+// fn.EventPublisher interface.
+var _ fn.EventPublisher[*AddrWithKeyInfo, QueryParams] = (*Book)(nil)
 
 // NewBook creates a new Book instance from the config.
 func NewBook(cfg BookConfig) *Book {
 	return &Book{
 		cfg: cfg,
 		subscribers: make(
-			map[uint64]*chanutils.EventReceiver[*AddrWithKeyInfo],
+			map[uint64]*fn.EventReceiver[*AddrWithKeyInfo],
 		),
 	}
 }
@@ -392,7 +392,7 @@ func (b *Book) CompleteEvent(ctx context.Context, event *Event,
 // marker onward existing items should be delivered on startup. If deliverFrom
 // is nil/zero/empty then all existing items will be delivered.
 func (b *Book) RegisterSubscriber(
-	receiver *chanutils.EventReceiver[*AddrWithKeyInfo],
+	receiver *fn.EventReceiver[*AddrWithKeyInfo],
 	deliverExisting bool, deliverFrom QueryParams) error {
 
 	b.subscriberMtx.Lock()
@@ -427,7 +427,7 @@ func (b *Book) RegisterSubscriber(
 // RemoveSubscriber removes the given subscriber and also stops it from
 // processing events.
 func (b *Book) RemoveSubscriber(
-	subscriber *chanutils.EventReceiver[*AddrWithKeyInfo]) error {
+	subscriber *fn.EventReceiver[*AddrWithKeyInfo]) error {
 
 	b.subscriberMtx.Lock()
 	defer b.subscriberMtx.Unlock()
diff --git a/cmd/tapd/main.go b/cmd/tapd/main.go
index abca0440..6e066619 100644
--- a/cmd/tapd/main.go
+++ b/cmd/tapd/main.go
@@ -7,7 +7,7 @@ import (
 	"runtime/pprof"
 
 	"github.com/jessevdk/go-flags"
-	"github.com/lightninglabs/taproot-assets/chanutils"
+	"github.com/lightninglabs/taproot-assets/fn"
 	"github.com/lightninglabs/taproot-assets/tapcfg"
 	"github.com/lightningnetwork/lnd/signal"
 )
@@ -62,9 +62,7 @@ func main() {
 	// This concurrent error queue can be used by every component that can
 	// raise runtime errors. Using a queue will prevent us from blocking on
 	// sending errors to it, as long as the queue is running.
-	errQueue := chanutils.NewConcurrentQueue[error](
-		chanutils.DefaultQueueSize,
-	)
+	errQueue := fn.NewConcurrentQueue[error](fn.DefaultQueueSize)
 	errQueue.Start()
 	defer errQueue.Stop()
 
diff --git a/commitment/asset.go b/commitment/asset.go
index 5d78f410..1722f45b 100644
--- a/commitment/asset.go
+++ b/commitment/asset.go
@@ -10,7 +10,7 @@ import (
 
 	"github.com/btcsuite/btcd/btcec/v2/schnorr"
 	"github.com/lightninglabs/taproot-assets/asset"
-	"github.com/lightninglabs/taproot-assets/chanutils"
+	"github.com/lightninglabs/taproot-assets/fn"
 	"github.com/lightninglabs/taproot-assets/mssmt"
 	"golang.org/x/exp/maps"
 )
@@ -404,7 +404,7 @@ func (c *AssetCommitment) Copy() (*AssetCommitment, error) {
 
 	// First, we'll perform a deep copy of all the assets that this existing
 	// commitment is committing to.
-	newAssets := chanutils.CopyAll(maps.Values(c.Assets()))
+	newAssets := fn.CopyAll(maps.Values(c.Assets()))
 
 	// Now that we have a deep copy of all the assets, we can just create a
 	// brand-new commitment from the set of assets.
diff --git a/commitment/proof.go b/commitment/proof.go
index 1eb4fb1f..c13e6571 100644
--- a/commitment/proof.go
+++ b/commitment/proof.go
@@ -5,7 +5,7 @@ import (
 	"io"
 
 	"github.com/lightninglabs/taproot-assets/asset"
-	"github.com/lightninglabs/taproot-assets/chanutils"
+	"github.com/lightninglabs/taproot-assets/fn"
 	"github.com/lightninglabs/taproot-assets/mssmt"
 	"github.com/lightningnetwork/lnd/tlv"
 )
@@ -129,7 +129,7 @@ func (p Proof) DeriveByAssetInclusion(asset *asset.Asset) (*TapCommitment,
 	)
 	log.Tracef("Derived asset inclusion proof for asset_id=%v, "+
 		"asset_commitment_key=%x, asset_commitment_leaf=%s",
-		asset.ID(), chanutils.ByteSlice(asset.AssetCommitmentKey()),
+		asset.ID(), fn.ByteSlice(asset.AssetCommitmentKey()),
 		assetCommitmentLeaf.NodeHash())
 
 	return NewTapCommitmentWithRoot(
diff --git a/commitment/tap.go b/commitment/tap.go
index 9ed2a152..18563716 100644
--- a/commitment/tap.go
+++ b/commitment/tap.go
@@ -11,7 +11,7 @@ import (
 	"github.com/btcsuite/btcd/chaincfg/chainhash"
 	"github.com/btcsuite/btcd/txscript"
 	"github.com/lightninglabs/taproot-assets/asset"
-	"github.com/lightninglabs/taproot-assets/chanutils"
+	"github.com/lightninglabs/taproot-assets/fn"
 	"github.com/lightninglabs/taproot-assets/mssmt"
 	"golang.org/x/exp/maps"
 )
@@ -378,7 +378,7 @@ func (c *TapCommitment) Copy() (*TapCommitment, error) {
 	}
 
 	// Otherwise, we'll copy all the internal asset commitments.
-	newAssetCommitments, err := chanutils.CopyAllErr(
+	newAssetCommitments, err := fn.CopyAllErr(
 		maps.Values(c.assetCommitments),
 	)
 	if err != nil {
diff --git a/chanutils/concurrency.go b/fn/concurrency.go
similarity index 98%
rename from chanutils/concurrency.go
rename to fn/concurrency.go
index 6dee610a..790f7514 100644
--- a/chanutils/concurrency.go
+++ b/fn/concurrency.go
@@ -1,4 +1,4 @@
-package chanutils
+package fn
 
 import (
 	"context"
diff --git a/chanutils/context_guard.go b/fn/context_guard.go
similarity index 99%
rename from chanutils/context_guard.go
rename to fn/context_guard.go
index f6bc50d7..a12e624f 100644
--- a/chanutils/context_guard.go
+++ b/fn/context_guard.go
@@ -1,4 +1,4 @@
-package chanutils
+package fn
 
 import (
 	"context"
diff --git a/chanutils/errors.go b/fn/errors.go
similarity index 97%
rename from chanutils/errors.go
rename to fn/errors.go
index 657326a4..9fdc785b 100644
--- a/chanutils/errors.go
+++ b/fn/errors.go
@@ -1,4 +1,4 @@
-package chanutils
+package fn
 
 import (
 	"context"
diff --git a/chanutils/events.go b/fn/events.go
similarity index 99%
rename from chanutils/events.go
rename to fn/events.go
index 9c55d13a..6ad18243 100644
--- a/chanutils/events.go
+++ b/fn/events.go
@@ -1,4 +1,4 @@
-package chanutils
+package fn
 
 import (
 	"fmt"
diff --git a/chanutils/func.go b/fn/func.go
similarity index 99%
rename from chanutils/func.go
rename to fn/func.go
index 3bb454c4..ed6e6942 100644
--- a/chanutils/func.go
+++ b/fn/func.go
@@ -1,4 +1,4 @@
-package chanutils
+package fn
 
 import "fmt"
 
diff --git a/chanutils/iter.go b/fn/iter.go
similarity index 98%
rename from chanutils/iter.go
rename to fn/iter.go
index 8ad876e9..1c9e81f3 100644
--- a/chanutils/iter.go
+++ b/fn/iter.go
@@ -1,4 +1,4 @@
-package chanutils
+package fn
 
 // ForEachErr will iterate through all items in the passed slice, calling the
 // function f on each slice. If a call to f fails, then the function returns an
diff --git a/chanutils/memory.go b/fn/memory.go
similarity index 98%
rename from chanutils/memory.go
rename to fn/memory.go
index 83b1d12b..0c71abc3 100644
--- a/chanutils/memory.go
+++ b/fn/memory.go
@@ -1,4 +1,4 @@
-package chanutils
+package fn
 
 // Ptr returns the pointer of the given value. This is useful in instances
 // where a function returns the value, but a pointer is wanted. Without this,
diff --git a/chanutils/queue.go b/fn/queue.go
similarity index 99%
rename from chanutils/queue.go
rename to fn/queue.go
index 7e043def..b337b358 100644
--- a/chanutils/queue.go
+++ b/fn/queue.go
@@ -1,4 +1,4 @@
-package chanutils
+package fn
 
 import (
 	"container/list"
diff --git a/chanutils/recv.go b/fn/recv.go
similarity index 98%
rename from chanutils/recv.go
rename to fn/recv.go
index 8fd43c8b..e4633e9e 100644
--- a/chanutils/recv.go
+++ b/fn/recv.go
@@ -1,4 +1,4 @@
-package chanutils
+package fn
 
 import (
 	"fmt"
diff --git a/chanutils/send.go b/fn/send.go
similarity index 96%
rename from chanutils/send.go
rename to fn/send.go
index ec84a3c5..6721362d 100644
--- a/chanutils/send.go
+++ b/fn/send.go
@@ -1,4 +1,4 @@
-package chanutils
+package fn
 
 // SendOrQuit attempts to and a message through channel c. If this succeeds,
 // then bool is returned. Otherwise if a quit signal is received first, then
diff --git a/chanutils/set.go b/fn/set.go
similarity index 99%
rename from chanutils/set.go
rename to fn/set.go
index de881610..51368b21 100644
--- a/chanutils/set.go
+++ b/fn/set.go
@@ -1,4 +1,4 @@
-package chanutils
+package fn
 
 import "golang.org/x/exp/maps"
 
diff --git a/mssmt/proof.go b/mssmt/proof.go
index b69e7c72..3114c5f3 100644
--- a/mssmt/proof.go
+++ b/mssmt/proof.go
@@ -4,7 +4,7 @@ import (
 	"errors"
 	"fmt"
 
-	"github.com/lightninglabs/taproot-assets/chanutils"
+	"github.com/lightninglabs/taproot-assets/fn"
 )
 
 var (
@@ -89,7 +89,7 @@ func (p *CompressedProof) Decompress() (*Proof, error) {
 	nodes := make([]Node, len(p.Bits))
 
 	// The number of 0 bits should match the number of pre-populated nodes.
-	numExpectedNodes := chanutils.Reduce(p.Bits, func(count int, bit bool) int {
+	numExpectedNodes := fn.Reduce(p.Bits, func(count int, bit bool) int {
 		if !bit {
 			return count + 1
 		}
diff --git a/proof/archive.go b/proof/archive.go
index 89c8e84a..35093c38 100644
--- a/proof/archive.go
+++ b/proof/archive.go
@@ -14,7 +14,7 @@ import (
 
 	"github.com/btcsuite/btcd/btcec/v2"
 	"github.com/lightninglabs/taproot-assets/asset"
-	"github.com/lightninglabs/taproot-assets/chanutils"
+	"github.com/lightninglabs/taproot-assets/fn"
 )
 
 const (
@@ -107,7 +107,7 @@ type Archiver interface {
 type NotifyArchiver interface {
 	Archiver
 
-	chanutils.EventPublisher[Blob, []*Locator]
+	fn.EventPublisher[Blob, []*Locator]
 }
 
 // FileArchiver implements proof Archiver backed by an on-disk file system. The
@@ -125,7 +125,7 @@ type FileArchiver struct {
 
 	// eventDistributor is an event distributor that will be used to notify
 	// subscribers about new proofs that are added to the archiver.
-	eventDistributor *chanutils.EventDistributor[Blob]
+	eventDistributor *fn.EventDistributor[Blob]
 }
 
 // NewFileArchiver creates a new file archive rooted at the passed specified
@@ -145,7 +145,7 @@ func NewFileArchiver(dirName string) (*FileArchiver, error) {
 
 	return &FileArchiver{
 		proofPath:        proofPath,
-		eventDistributor: chanutils.NewEventDistributor[Blob](),
+		eventDistributor: fn.NewEventDistributor[Blob](),
 	}, nil
 }
 
@@ -232,7 +232,7 @@ func (f *FileArchiver) ImportProofs(_ context.Context,
 // marker onward existing items should be delivered on startup. If deliverFrom
 // is nil/zero/empty then all existing items will be delivered.
 func (f *FileArchiver) RegisterSubscriber(
-	receiver *chanutils.EventReceiver[Blob],
+	receiver *fn.EventReceiver[Blob],
 	deliverExisting bool, deliverFrom []*Locator) error {
 
 	f.eventDistributor.RegisterSubscriber(receiver)
@@ -259,7 +259,7 @@ func (f *FileArchiver) RegisterSubscriber(
 // RemoveSubscriber removes the given subscriber and also stops it from
 // processing events.
 func (f *FileArchiver) RemoveSubscriber(
-	subscriber *chanutils.EventReceiver[Blob]) error {
+	subscriber *fn.EventReceiver[Blob]) error {
 
 	return f.eventDistributor.RemoveSubscriber(subscriber)
 }
@@ -281,7 +281,7 @@ type MultiArchiver struct {
 
 	// eventDistributor is an event distributor that will be used to notify
 	// subscribers about new proofs that are added to the archiver.
-	eventDistributor *chanutils.EventDistributor[Blob]
+	eventDistributor *fn.EventDistributor[Blob]
 }
 
 // NewMultiArchiver creates a new MultiArchiver based on the set of specified
@@ -293,7 +293,7 @@ func NewMultiArchiver(verifier Verifier, archiveTimeout time.Duration,
 		proofVerifier:    verifier,
 		backends:         backends,
 		archiveTimeout:   archiveTimeout,
-		eventDistributor: chanutils.NewEventDistributor[Blob](),
+		eventDistributor: fn.NewEventDistributor[Blob](),
 	}
 }
 
@@ -366,7 +366,7 @@ func (m *MultiArchiver) ImportProofs(ctx context.Context,
 		return nil
 	}
 
-	if err := chanutils.ParSlice(ctx, proofs, f); err != nil {
+	if err := fn.ParSlice(ctx, proofs, f); err != nil {
 		return err
 	}
 
@@ -381,7 +381,7 @@ func (m *MultiArchiver) ImportProofs(ctx context.Context,
 	}
 
 	// Deliver each new proof to the new item queue of the subscribers.
-	blobs := chanutils.Map(proofs, func(p *AnnotatedProof) Blob {
+	blobs := fn.Map(proofs, func(p *AnnotatedProof) Blob {
 		return p.Blob
 	})
 	m.eventDistributor.NotifySubscribers(blobs...)
@@ -395,8 +395,7 @@ func (m *MultiArchiver) ImportProofs(ctx context.Context,
 // optional deliverFrom can be specified to indicate from which timestamp/index/
 // marker onward existing items should be delivered on startup. If deliverFrom
 // is nil/zero/empty then all existing items will be delivered.
-func (m *MultiArchiver) RegisterSubscriber(
-	receiver *chanutils.EventReceiver[Blob],
+func (m *MultiArchiver) RegisterSubscriber(receiver *fn.EventReceiver[Blob],
 	deliverExisting bool, deliverFrom []*Locator) error {
 
 	m.eventDistributor.RegisterSubscriber(receiver)
@@ -428,7 +427,7 @@ func (m *MultiArchiver) RegisterSubscriber(
 // RemoveSubscriber removes the given subscriber and also stops it from
 // processing events.
 func (m *MultiArchiver) RemoveSubscriber(
-	subscriber *chanutils.EventReceiver[Blob]) error {
+	subscriber *fn.EventReceiver[Blob]) error {
 
 	return m.eventDistributor.RemoveSubscriber(subscriber)
 }
diff --git a/proof/courier.go b/proof/courier.go
index a2d51e03..aa6086e6 100644
--- a/proof/courier.go
+++ b/proof/courier.go
@@ -12,7 +12,7 @@ import (
 	"github.com/btcsuite/btcd/btcec/v2"
 	"github.com/lightninglabs/lightning-node-connect/hashmailrpc"
 	"github.com/lightninglabs/taproot-assets/asset"
-	"github.com/lightninglabs/taproot-assets/chanutils"
+	"github.com/lightninglabs/taproot-assets/fn"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/credentials"
@@ -60,7 +60,7 @@ type Courier[Addr any] interface {
 
 	// SetSubscribers sets the set of subscribers that will be notified
 	// of proof courier related events.
-	SetSubscribers(map[uint64]*chanutils.EventReceiver[chanutils.Event])
+	SetSubscribers(map[uint64]*fn.EventReceiver[fn.Event])
 }
 
 // ProofMailbox represents an abstract store-and-forward mailbox that can be
@@ -369,7 +369,7 @@ type HashMailCourier struct {
 
 	// subscribers is a map of components that want to be notified on new
 	// events, keyed by their subscription ID.
-	subscribers map[uint64]*chanutils.EventReceiver[chanutils.Event]
+	subscribers map[uint64]*fn.EventReceiver[fn.Event]
 
 	// subscriberMtx guards the subscribers map and access to the
 	// subscriptionID.
@@ -383,7 +383,7 @@ func NewHashMailCourier(cfg *HashMailCourierCfg, mailbox ProofMailbox,
 	deliveryLog DeliveryLog) (*HashMailCourier, error) {
 
 	subscribers := make(
-		map[uint64]*chanutils.EventReceiver[chanutils.Event],
+		map[uint64]*fn.EventReceiver[fn.Event],
 	)
 	return &HashMailCourier{
 		cfg:         cfg,
@@ -638,7 +638,7 @@ func (h *HashMailCourier) backoffExec(ctx context.Context,
 }
 
 // publishSubscriberEvent publishes an event to all subscribers.
-func (h *HashMailCourier) publishSubscriberEvent(event chanutils.Event) {
+func (h *HashMailCourier) publishSubscriberEvent(event fn.Event) {
 	// Lock the subscriber mutex to ensure that we don't modify the
 	// subscriber map while we're iterating over it.
 	h.subscriberMtx.Lock()
@@ -733,7 +733,7 @@ func (h *HashMailCourier) ReceiveProof(ctx context.Context, recipient Recipient,
 // SetSubscribers sets the subscribers for the courier. This method is
 // thread-safe.
 func (h *HashMailCourier) SetSubscribers(
-	subscribers map[uint64]*chanutils.EventReceiver[chanutils.Event]) {
+	subscribers map[uint64]*fn.EventReceiver[fn.Event]) {
 
 	h.subscriberMtx.Lock()
 	defer h.subscriberMtx.Unlock()
diff --git a/proof/taproot.go b/proof/taproot.go
index e3993a88..5c79da31 100644
--- a/proof/taproot.go
+++ b/proof/taproot.go
@@ -12,8 +12,8 @@ import (
 	"github.com/btcsuite/btcd/txscript"
 	"github.com/davecgh/go-spew/spew"
 	"github.com/lightninglabs/taproot-assets/asset"
-	"github.com/lightninglabs/taproot-assets/chanutils"
 	"github.com/lightninglabs/taproot-assets/commitment"
+	"github.com/lightninglabs/taproot-assets/fn"
 	"github.com/lightningnetwork/lnd/tlv"
 )
 
@@ -291,7 +291,7 @@ func (p TaprootProof) DeriveByAssetInclusion(
 
 	log.Tracef("Derived Taproot Asset commitment taproot_asset_root=%x, "+
 		"internal_key=%x, taproot_key=%x",
-		chanutils.ByteSlice(tapCommitment.TapscriptRoot(nil)),
+		fn.ByteSlice(tapCommitment.TapscriptRoot(nil)),
 		p.InternalKey.SerializeCompressed(),
 		schnorr.SerializePubKey(pubKey))
 
@@ -349,7 +349,7 @@ func (p TaprootProof) DeriveByAssetExclusion(assetCommitmentKey,
 
 	log.Tracef("Derived Taproot Asset commitment taproot_asset_root=%x, "+
 		"internal_key=%x",
-		chanutils.ByteSlice(commitment.TapscriptRoot(nil)),
+		fn.ByteSlice(commitment.TapscriptRoot(nil)),
 		p.InternalKey.SerializeCompressed())
 
 	return deriveTaprootKeysFromTapCommitment(
diff --git a/rpcserver.go b/rpcserver.go
index fdc86ef9..aa49cb03 100644
--- a/rpcserver.go
+++ b/rpcserver.go
@@ -22,8 +22,8 @@ import (
 	proxy "github.com/grpc-ecosystem/grpc-gateway/v2/runtime"
 	"github.com/lightninglabs/taproot-assets/address"
 	"github.com/lightninglabs/taproot-assets/asset"
-	"github.com/lightninglabs/taproot-assets/chanutils"
 	"github.com/lightninglabs/taproot-assets/commitment"
+	"github.com/lightninglabs/taproot-assets/fn"
 	"github.com/lightninglabs/taproot-assets/mssmt"
 	"github.com/lightninglabs/taproot-assets/proof"
 	"github.com/lightninglabs/taproot-assets/rpcperms"
@@ -435,7 +435,7 @@ func (r *rpcServer) ListBatches(_ context.Context,
 		return nil, fmt.Errorf("unable to list batches: %w", err)
 	}
 
-	rpcBatches, err := chanutils.MapErr(batches, marshalMintingBatch)
+	rpcBatches, err := fn.MapErr(batches, marshalMintingBatch)
 	if err != nil {
 		return nil, err
 	}
@@ -1760,9 +1760,7 @@ func (r *rpcServer) SubscribeSendAssetEventNtfns(
 
 	// Create a new event subscriber and pass a copy to the chain porter.
 	// We will then read events from the subscriber.
-	eventSubscriber := chanutils.NewEventReceiver[chanutils.Event](
-		chanutils.DefaultQueueSize,
-	)
+	eventSubscriber := fn.NewEventReceiver[fn.Event](fn.DefaultQueueSize)
 	defer eventSubscriber.Stop()
 
 	err := r.cfg.ChainPorter.RegisterSubscriber(eventSubscriber, false, false)
@@ -1816,7 +1814,7 @@ func (r *rpcServer) SubscribeSendAssetEventNtfns(
 
 // marshallSendAssetEvent maps a ChainPorter event to its RPC counterpart.
 func marshallSendAssetEvent(
-	eventInterface chanutils.Event) (*taprpc.SendAssetEvent, error) {
+	eventInterface fn.Event) (*taprpc.SendAssetEvent, error) {
 
 	switch event := eventInterface.(type) {
 	case *tapfreighter.ExecuteSendStateEvent:
@@ -1863,7 +1861,7 @@ func marshalMintingBatch(batch *tapgarden.MintingBatch) (*mintrpc.MintingBatch,
 		var seedlingMeta *taprpc.AssetMeta
 		if seedling.Meta != nil {
 			seedlingMeta = &taprpc.AssetMeta{
-				MetaHash: chanutils.ByteSlice(
+				MetaHash: fn.ByteSlice(
 					seedling.Meta.MetaHash(),
 				),
 				Data: seedling.Meta.Data,
@@ -2596,15 +2594,15 @@ func (r *rpcServer) marshalUniverseDiff(ctx context.Context,
 		SyncedUniverses: make([]*unirpc.SyncedUniverse, 0, len(uniDiff)),
 	}
 
-	for _, diff := range uniDiff {
+	err := fn.ForEachErr(uniDiff, func(diff universe.AssetSyncDiff) error {
 		oldUniRoot, err := marshalUniverseRoot(diff.OldUniverseRoot)
 		if err != nil {
-			return nil, fmt.Errorf("unable to marshal old "+
-				"uni root: %w", err)
+			return fmt.Errorf("unable to marshal old uni "+
+				"root: %w", err)
 		}
 		newUniRoot, err := marshalUniverseRoot(diff.NewUniverseRoot)
 		if err != nil {
-			return nil, fmt.Errorf("unable to marshal new unit "+
+			return fmt.Errorf("unable to marshal new unit "+
 				"root: %w", err)
 		}
 
@@ -2612,7 +2610,7 @@ func (r *rpcServer) marshalUniverseDiff(ctx context.Context,
 		for i, leaf := range diff.NewLeafProofs {
 			leaves[i], err = r.marshalAssetLeaf(ctx, leaf)
 			if err != nil {
-				return nil, err
+				return err
 			}
 		}
 
@@ -2623,6 +2621,10 @@ func (r *rpcServer) marshalUniverseDiff(ctx context.Context,
 				NewAssetLeaves: leaves,
 			},
 		)
+		return nil
+	})
+	if err != nil {
+		return nil, err
 	}
 
 	return resp, nil
@@ -2684,7 +2686,7 @@ func (r *rpcServer) ListFederationServers(ctx context.Context,
 	}
 
 	return &unirpc.ListFederationServersResponse{
-		Servers: chanutils.Map(uniServers, marshalUniverseServer),
+		Servers: fn.Map(uniServers, marshalUniverseServer),
 	}, nil
 }
 
@@ -2701,7 +2703,7 @@ func (r *rpcServer) AddFederationServer(ctx context.Context,
 	in *unirpc.AddFederationServerRequest,
 ) (*unirpc.AddFederationServerResponse, error) {
 
-	serversToAdd := chanutils.Map(in.Servers, unmarshalUniverseServer)
+	serversToAdd := fn.Map(in.Servers, unmarshalUniverseServer)
 
 	err := r.cfg.UniverseFederation.AddServer(serversToAdd...)
 	if err != nil {
@@ -2717,7 +2719,7 @@ func (r *rpcServer) DeleteFederationServer(ctx context.Context,
 	in *unirpc.DeleteFederationServerRequest,
 ) (*unirpc.DeleteFederationServerResponse, error) {
 
-	serversToDel := chanutils.Map(in.Servers, unmarshalUniverseServer)
+	serversToDel := fn.Map(in.Servers, unmarshalUniverseServer)
 
 	err := r.cfg.FederationDB.RemoveServers(ctx, serversToDel...)
 	if err != nil {
@@ -2748,7 +2750,7 @@ func (r *rpcServer) ProveAssetOwnership(ctx context.Context,
 		return nil, fmt.Errorf("asset ID must be 32 bytes")
 	}
 
-	assetID := chanutils.ToArray[asset.ID](in.AssetId)
+	assetID := fn.ToArray[asset.ID](in.AssetId)
 	proofBlob, err := r.cfg.ProofArchive.FetchProof(ctx, proof.Locator{
 		AssetID:   &assetID,
 		ScriptKey: *scriptKey,
@@ -2875,16 +2877,16 @@ func (r *rpcServer) QueryAssetStats(ctx context.Context,
 			AssetTypeFilter: func() *asset.Type {
 				switch req.AssetTypeFilter {
 				case unirpc.AssetTypeFilter_FILTER_ASSET_NORMAL:
-					return chanutils.Ptr(asset.Normal)
+					return fn.Ptr(asset.Normal)
 
 				case unirpc.AssetTypeFilter_FILTER_ASSET_COLLECTIBLE:
-					return chanutils.Ptr(asset.Collectible)
+					return fn.Ptr(asset.Collectible)
 
 				default:
 					return nil
 				}
 			}(),
-			AssetIDFilter: chanutils.ToArray[asset.ID](
+			AssetIDFilter: fn.ToArray[asset.ID](
 				req.AssetIdFilter,
 			),
 			SortBy: universe.SyncStatsSort(req.SortBy),
@@ -2898,7 +2900,7 @@ func (r *rpcServer) QueryAssetStats(ctx context.Context,
 
 	snapshots := assetStats.SyncStats
 	resp := &unirpc.UniverseAssetStats{
-		AssetStats: chanutils.Map(snapshots, marshalAssetSyncSnapshot),
+		AssetStats: fn.Map(snapshots, marshalAssetSyncSnapshot),
 	}
 
 	return resp, nil
diff --git a/server.go b/server.go
index 29805087..aa7f7f81 100644
--- a/server.go
+++ b/server.go
@@ -10,7 +10,7 @@ import (
 
 	proxy "github.com/grpc-ecosystem/grpc-gateway/v2/runtime"
 	"github.com/lightninglabs/lndclient"
-	"github.com/lightninglabs/taproot-assets/chanutils"
+	"github.com/lightninglabs/taproot-assets/fn"
 	"github.com/lightninglabs/taproot-assets/perms"
 	"github.com/lightninglabs/taproot-assets/rpcperms"
 	"github.com/lightninglabs/taproot-assets/taprpc"
@@ -301,7 +301,7 @@ func (s *Server) RunUntilShutdown(mainErrChan <-chan error) error {
 
 		// We'll report the error to the main daemon, but only if this
 		// isn't a context cancel.
-		if chanutils.IsCanceled(err) {
+		if fn.IsCanceled(err) {
 			return nil
 		}
 
diff --git a/tapdb/addrs.go b/tapdb/addrs.go
index 8e61e41a..7b577be6 100644
--- a/tapdb/addrs.go
+++ b/tapdb/addrs.go
@@ -17,8 +17,8 @@ import (
 	"github.com/lightninglabs/lndclient"
 	"github.com/lightninglabs/taproot-assets/address"
 	"github.com/lightninglabs/taproot-assets/asset"
-	"github.com/lightninglabs/taproot-assets/chanutils"
 	"github.com/lightninglabs/taproot-assets/commitment"
+	"github.com/lightninglabs/taproot-assets/fn"
 	"github.com/lightninglabs/taproot-assets/tapdb/sqlc"
 	"github.com/lightningnetwork/lnd/keychain"
 )
@@ -901,7 +901,7 @@ func (a *TapAddressBook) QueryAssetGroup(ctx context.Context,
 		assetGroup.Genesis = &asset.Genesis{
 			FirstPrevOut: genesisPrevOut,
 			Tag:          assetGen.AssetTag,
-			MetaHash: chanutils.ToArray[[32]byte](
+			MetaHash: fn.ToArray[[32]byte](
 				assetGen.MetaHash,
 			),
 			OutputIndex: uint32(assetGen.OutputIndex),
diff --git a/tapdb/asset_minting.go b/tapdb/asset_minting.go
index 02ba42e7..6cd82c94 100644
--- a/tapdb/asset_minting.go
+++ b/tapdb/asset_minting.go
@@ -13,8 +13,8 @@ import (
 	"github.com/btcsuite/btcd/chaincfg/chainhash"
 	"github.com/btcsuite/btcd/wire"
 	"github.com/lightninglabs/taproot-assets/asset"
-	"github.com/lightninglabs/taproot-assets/chanutils"
 	"github.com/lightninglabs/taproot-assets/commitment"
+	"github.com/lightninglabs/taproot-assets/fn"
 	"github.com/lightninglabs/taproot-assets/proof"
 	"github.com/lightninglabs/taproot-assets/tapdb/sqlc"
 	"github.com/lightninglabs/taproot-assets/tapgarden"
@@ -723,7 +723,7 @@ func (a *AssetMintingStore) FetchNonFinalBatches(
 			return marshalMintingBatch(ctx, q, convBatch)
 		}
 
-		batches, err = chanutils.MapErr(dbBatches, parseBatch)
+		batches, err = fn.MapErr(dbBatches, parseBatch)
 		if err != nil {
 			return fmt.Errorf("batch parsing failed: %w", err)
 		}
@@ -758,7 +758,7 @@ func (a *AssetMintingStore) FetchAllBatches(
 			return marshalMintingBatch(ctx, q, convBatch)
 		}
 
-		batches, err = chanutils.MapErr(dbBatches, parseBatch)
+		batches, err = fn.MapErr(dbBatches, parseBatch)
 		if err != nil {
 			return fmt.Errorf("batch parsing failed: %w", err)
 		}
diff --git a/tapdb/assets_store.go b/tapdb/assets_store.go
index c87abddd..876f8cf4 100644
--- a/tapdb/assets_store.go
+++ b/tapdb/assets_store.go
@@ -15,8 +15,8 @@ import (
 	"github.com/btcsuite/btcd/chaincfg/chainhash"
 	"github.com/btcsuite/btcd/wire"
 	"github.com/lightninglabs/taproot-assets/asset"
-	"github.com/lightninglabs/taproot-assets/chanutils"
 	"github.com/lightninglabs/taproot-assets/commitment"
+	"github.com/lightninglabs/taproot-assets/fn"
 	"github.com/lightninglabs/taproot-assets/mssmt"
 	"github.com/lightninglabs/taproot-assets/proof"
 	"github.com/lightninglabs/taproot-assets/tapdb/sqlc"
@@ -308,7 +308,7 @@ type AssetStore struct {
 
 	// eventDistributor is an event distributor that will be used to notify
 	// subscribers about new proofs that are added to the archiver.
-	eventDistributor *chanutils.EventDistributor[proof.Blob]
+	eventDistributor *fn.EventDistributor[proof.Blob]
 }
 
 // NewAssetStore creates a new AssetStore from the specified BatchedAssetStore
@@ -316,7 +316,7 @@ type AssetStore struct {
 func NewAssetStore(db BatchedAssetStore) *AssetStore {
 	return &AssetStore{
 		db:               db,
-		eventDistributor: chanutils.NewEventDistributor[proof.Blob](),
+		eventDistributor: fn.NewEventDistributor[proof.Blob](),
 	}
 }
 
@@ -1353,11 +1353,9 @@ func (a *AssetStore) ImportProofs(ctx context.Context,
 	// Notify any event subscribers that there are new proofs. We do this
 	// outside of the transaction to avoid the subscribers trying to look up
 	// the proofs before they are committed.
-	proofBlobs := chanutils.Map(
-		proofs, func(p *proof.AnnotatedProof) proof.Blob {
-			return p.Blob
-		},
-	)
+	proofBlobs := fn.Map(proofs, func(p *proof.AnnotatedProof) proof.Blob {
+		return p.Blob
+	})
 	a.eventDistributor.NotifySubscribers(proofBlobs...)
 
 	return nil
@@ -1370,7 +1368,7 @@ func (a *AssetStore) ImportProofs(ctx context.Context,
 // marker onward existing items should be delivered on startup. If deliverFrom
 // is nil/zero/empty then all existing items will be delivered.
 func (a *AssetStore) RegisterSubscriber(
-	receiver *chanutils.EventReceiver[proof.Blob],
+	receiver *fn.EventReceiver[proof.Blob],
 	deliverExisting bool, deliverFrom []*proof.Locator) error {
 
 	a.eventDistributor.RegisterSubscriber(receiver)
@@ -1398,7 +1396,7 @@ func (a *AssetStore) RegisterSubscriber(
 // RemoveSubscriber removes the given subscriber and also stops it from
 // processing events.
 func (a *AssetStore) RemoveSubscriber(
-	subscriber *chanutils.EventReceiver[proof.Blob]) error {
+	subscriber *fn.EventReceiver[proof.Blob]) error {
 
 	return a.eventDistributor.RemoveSubscriber(subscriber)
 }
@@ -1563,10 +1561,8 @@ func (a *AssetStore) queryCommitments(ctx context.Context,
 			return cAsset.Asset
 		}
 
-		assets := chanutils.Map(anchoredAssets, fetchAsset)
-
+		assets := fn.Map(anchoredAssets, fetchAsset)
 		tapCommitment, err := commitment.FromAssets(assets...)
-
 		if err != nil {
 			return nil, err
 		}
diff --git a/tapdb/universe.go b/tapdb/universe.go
index c03c46c7..47230007 100644
--- a/tapdb/universe.go
+++ b/tapdb/universe.go
@@ -12,7 +12,7 @@ import (
 	"github.com/btcsuite/btcd/btcec/v2/schnorr"
 	"github.com/btcsuite/btcd/wire"
 	"github.com/lightninglabs/taproot-assets/asset"
-	"github.com/lightninglabs/taproot-assets/chanutils"
+	"github.com/lightninglabs/taproot-assets/fn"
 	"github.com/lightninglabs/taproot-assets/mssmt"
 	"github.com/lightninglabs/taproot-assets/proof"
 	"github.com/lightninglabs/taproot-assets/tapdb/sqlc"
@@ -308,7 +308,7 @@ func (b *BaseUniverseTree) RegisterIssuance(ctx context.Context,
 		// overlay.
 		universeRootID, err := db.UpsertUniverseRoot(ctx, NewUniverseRoot{
 			NamespaceRoot: b.smtNamespace,
-			AssetID:       chanutils.ByteSlice(leaf.ID()),
+			AssetID:       fn.ByteSlice(leaf.ID()),
 			GroupKey:      groupKeyBytes,
 		})
 		if err != nil {
@@ -437,7 +437,7 @@ func (b *BaseUniverseTree) FetchIssuanceProof(ctx context.Context,
 		// Now that we have all the leaves we need to query, we'll look
 		// each up them up in the universe tree, obtaining a merkle
 		// proof for each of them along the way.
-		for _, leaf := range universeLeaves {
+		return fn.ForEachErr(universeLeaves, func(leaf UniverseLeaf) error {
 			scriptPub, err := schnorr.ParsePubKey(leaf.ScriptKeyBytes)
 			if err != nil {
 				return err
@@ -493,9 +493,9 @@ func (b *BaseUniverseTree) FetchIssuanceProof(ctx context.Context,
 			}
 
 			proofs = append(proofs, proof)
-		}
 
-		return nil
+			return nil
+		})
 	})
 	if dbErr != nil {
 		return nil, dbErr
@@ -517,7 +517,7 @@ func (b *BaseUniverseTree) MintingKeys(ctx context.Context,
 			return err
 		}
 
-		for _, key := range universeKeys {
+		return fn.ForEachErr(universeKeys, func(key UniverseKeys) error {
 			scriptKeyPub, err := schnorr.ParsePubKey(
 				key.ScriptKeyBytes,
 			)
@@ -539,9 +539,9 @@ func (b *BaseUniverseTree) MintingKeys(ctx context.Context,
 				MintingOutpoint: genPoint,
 				ScriptKey:       &scriptKey,
 			})
-		}
 
-		return nil
+			return nil
+		})
 	})
 	if dbErr != nil {
 		return nil, dbErr
@@ -571,11 +571,11 @@ func (b *BaseUniverseTree) MintingLeaves(ctx context.Context,
 			return err
 		}
 
-		for _, leaf := range universeLeaves {
+		return fn.ForEachErr(universeLeaves, func(dbLeaf UniverseLeaf) error {
 			// For each leaf, we'll decode the proof, and then also
 			// fetch the genesis asset information for that leaf.
 			leafAssetGen, err := fetchGenesis(
-				ctx, db, leaf.GenAssetID,
+				ctx, db, dbLeaf.GenAssetID,
 			)
 			if err != nil {
 				return err
@@ -587,8 +587,8 @@ func (b *BaseUniverseTree) MintingLeaves(ctx context.Context,
 				GenesisWithGroup: universe.GenesisWithGroup{
 					Genesis: leafAssetGen,
 				},
-				GenesisProof: leaf.GenesisProof,
-				Amt:          uint64(leaf.SumAmt),
+				GenesisProof: dbLeaf.GenesisProof,
+				Amt:          uint64(dbLeaf.SumAmt),
 			}
 			if b.id.GroupKey != nil {
 				leaf.GroupKey = &asset.GroupKey{
@@ -597,9 +597,9 @@ func (b *BaseUniverseTree) MintingLeaves(ctx context.Context,
 			}
 
 			leaves = append(leaves, leaf)
-		}
 
-		return nil
+			return nil
+		})
 	})
 	if dbErr != nil {
 		return nil, dbErr
diff --git a/tapdb/universe_federation.go b/tapdb/universe_federation.go
index b3f19b0b..25da9f6c 100644
--- a/tapdb/universe_federation.go
+++ b/tapdb/universe_federation.go
@@ -6,7 +6,7 @@ import (
 	"fmt"
 	"time"
 
-	"github.com/lightninglabs/taproot-assets/chanutils"
+	"github.com/lightninglabs/taproot-assets/fn"
 	"github.com/lightninglabs/taproot-assets/tapdb/sqlc"
 	"github.com/lightninglabs/taproot-assets/universe"
 )
@@ -89,7 +89,7 @@ func (u *UniverseFederationDB) UniverseServers(ctx context.Context,
 			return err
 		}
 
-		uniServers = chanutils.Map(servers,
+		uniServers = fn.Map(servers,
 			func(s sqlc.UniverseServer) universe.ServerAddr {
 				return universe.NewServerAddr(
 					uint32(s.ID), s.ServerHost,
@@ -109,14 +109,13 @@ func (u *UniverseFederationDB) AddServers(ctx context.Context,
 
 	var writeTx UniverseFederationOptions
 	err := u.db.ExecTx(ctx, &writeTx, func(db UniverseServerStore) error {
-		return chanutils.ForEachErr(addrs,
-			func(a universe.ServerAddr) error {
-				addr := NewUniverseServer{
-					ServerHost:   a.HostStr(),
-					LastSyncTime: time.Now(),
-				}
-				return db.InsertUniverseServer(ctx, addr)
-			})
+		return fn.ForEachErr(addrs, func(a universe.ServerAddr) error {
+			addr := NewUniverseServer{
+				ServerHost:   a.HostStr(),
+				LastSyncTime: time.Now(),
+			}
+			return db.InsertUniverseServer(ctx, addr)
+		})
 	})
 	if err != nil {
 		// Add context to unique constraint errors.
@@ -138,7 +137,7 @@ func (u *UniverseFederationDB) RemoveServers(ctx context.Context,
 
 	var writeTx UniverseFederationOptions
 	return u.db.ExecTx(ctx, &writeTx, func(db UniverseServerStore) error {
-		return chanutils.ForEachErr(addrs, func(a universe.ServerAddr) error {
+		return fn.ForEachErr(addrs, func(a universe.ServerAddr) error {
 			// If the host string is set, then we'll make the
 			// target ID -1 so we can target _only_ based on the
 			// host string instead. This avoids bugs where a user
@@ -164,7 +163,7 @@ func (u *UniverseFederationDB) LogNewSyncs(ctx context.Context,
 
 	var writeTx UniverseFederationOptions
 	return u.db.ExecTx(ctx, &writeTx, func(db UniverseServerStore) error {
-		return chanutils.ForEachErr(addrs, func(a universe.ServerAddr) error {
+		return fn.ForEachErr(addrs, func(a universe.ServerAddr) error {
 			return db.LogServerSync(ctx, sqlc.LogServerSyncParams{
 				NewSyncTime:  time.Now(),
 				TargetServer: a.HostStr(),
diff --git a/tapdb/universe_stats.go b/tapdb/universe_stats.go
index 37fc5921..d8ad9cbf 100644
--- a/tapdb/universe_stats.go
+++ b/tapdb/universe_stats.go
@@ -9,7 +9,7 @@ import (
 	"time"
 
 	"github.com/lightninglabs/taproot-assets/asset"
-	"github.com/lightninglabs/taproot-assets/chanutils"
+	"github.com/lightninglabs/taproot-assets/fn"
 	"github.com/lightninglabs/taproot-assets/tapdb/sqlc"
 	"github.com/lightninglabs/taproot-assets/universe"
 )
@@ -251,7 +251,7 @@ func (u *UniverseStats) QuerySyncStats(ctx context.Context,
 		for _, assetStat := range assetStats {
 			stats := universe.AssetSyncSnapshot{
 				TotalSupply: uint64(assetStat.AssetSupply),
-				AssetID: chanutils.ToArray[asset.ID](
+				AssetID: fn.ToArray[asset.ID](
 					assetStat.AssetID,
 				),
 				AssetName:   assetStat.AssetName,
diff --git a/tapfreighter/chain_porter.go b/tapfreighter/chain_porter.go
index b8b59a96..a175a51d 100644
--- a/tapfreighter/chain_porter.go
+++ b/tapfreighter/chain_porter.go
@@ -14,7 +14,7 @@ import (
 	"github.com/btcsuite/btcd/txscript"
 	"github.com/btcsuite/btcd/wire"
 	"github.com/lightninglabs/taproot-assets/asset"
-	"github.com/lightninglabs/taproot-assets/chanutils"
+	"github.com/lightninglabs/taproot-assets/fn"
 	"github.com/lightninglabs/taproot-assets/proof"
 	"github.com/lightninglabs/taproot-assets/tapgarden"
 	"github.com/lightninglabs/taproot-assets/tappsbt"
@@ -83,26 +83,26 @@ type ChainPorter struct {
 
 	// subscribers is a map of components that want to be notified on new
 	// events, keyed by their subscription ID.
-	subscribers map[uint64]*chanutils.EventReceiver[chanutils.Event]
+	subscribers map[uint64]*fn.EventReceiver[fn.Event]
 
 	// subscriberMtx guards the subscribers map and access to the
 	// subscriptionID.
 	subscriberMtx sync.Mutex
 
-	*chanutils.ContextGuard
+	*fn.ContextGuard
 }
 
 // NewChainPorter creates a new instance of the ChainPorter given a valid
 // config.
 func NewChainPorter(cfg *ChainPorterConfig) *ChainPorter {
 	subscribers := make(
-		map[uint64]*chanutils.EventReceiver[chanutils.Event],
+		map[uint64]*fn.EventReceiver[fn.Event],
 	)
 	return &ChainPorter{
 		cfg:         cfg,
 		exportReqs:  make(chan Parcel),
 		subscribers: subscribers,
-		ContextGuard: &chanutils.ContextGuard{
+		ContextGuard: &fn.ContextGuard{
 			DefaultTimeout: tapgarden.DefaultTimeout,
 			Quit:           make(chan struct{}),
 		},
@@ -169,7 +169,7 @@ func (p *ChainPorter) Stop() error {
 // RequestShipment is the main external entry point to the porter. This request
 // a new transfer take place.
 func (p *ChainPorter) RequestShipment(req Parcel) (*OutboundParcel, error) {
-	if !chanutils.SendOrQuit(p.exportReqs, req, p.Quit) {
+	if !fn.SendOrQuit(p.exportReqs, req, p.Quit) {
 		return nil, fmt.Errorf("ChainPorter shutting down")
 	}
 
@@ -593,7 +593,7 @@ func (p *ChainPorter) transferReceiverProof(pkg *sendPackage) error {
 		ctx, cancel := p.WithCtxQuitNoTimeout()
 		defer cancel()
 
-		err := chanutils.ParSlice(ctx, pkg.OutboundPkg.Outputs, deliver)
+		err := fn.ParSlice(ctx, pkg.OutboundPkg.Outputs, deliver)
 		if err != nil {
 			return fmt.Errorf("error delivering proof(s): %w", err)
 		}
@@ -992,7 +992,7 @@ func (p *ChainPorter) stateStep(currentPkg sendPackage) (*sendPackage, error) {
 //
 // TODO(ffranr): Add support for delivering existing events to new subscribers.
 func (p *ChainPorter) RegisterSubscriber(
-	receiver *chanutils.EventReceiver[chanutils.Event],
+	receiver *fn.EventReceiver[fn.Event],
 	deliverExisting bool, deliverFrom bool) error {
 
 	p.subscriberMtx.Lock()
@@ -1011,7 +1011,7 @@ func (p *ChainPorter) RegisterSubscriber(
 // RemoveSubscriber removes a subscriber from the set of subscribers that will
 // be notified of any new events that are broadcast.
 func (p *ChainPorter) RemoveSubscriber(
-	subscriber *chanutils.EventReceiver[chanutils.Event]) error {
+	subscriber *fn.EventReceiver[fn.Event]) error {
 
 	p.subscriberMtx.Lock()
 	defer p.subscriberMtx.Unlock()
@@ -1034,7 +1034,7 @@ func (p *ChainPorter) RemoveSubscriber(
 }
 
 // publishSubscriberEvent publishes an event to all subscribers.
-func (p *ChainPorter) publishSubscriberEvent(event chanutils.Event) {
+func (p *ChainPorter) publishSubscriberEvent(event fn.Event) {
 	// Lock the subscriber mutex to ensure that we don't modify the
 	// subscriber map while we're iterating over it.
 	p.subscriberMtx.Lock()
@@ -1046,8 +1046,8 @@ func (p *ChainPorter) publishSubscriberEvent(event chanutils.Event) {
 }
 
 // A compile-time assertion to make sure ChainPorter satisfies the
-// chanutils.EventPublisher interface.
-var _ chanutils.EventPublisher[chanutils.Event, bool] = (*ChainPorter)(nil)
+// fn.EventPublisher interface.
+var _ fn.EventPublisher[fn.Event, bool] = (*ChainPorter)(nil)
 
 // ExecuteSendStateEvent is an event which is sent to the ChainPorter's event
 // subscribers before a state is executed.
diff --git a/tapfreighter/interface.go b/tapfreighter/interface.go
index b34b15ca..f8272e7d 100644
--- a/tapfreighter/interface.go
+++ b/tapfreighter/interface.go
@@ -11,8 +11,8 @@ import (
 	"github.com/btcsuite/btcd/chaincfg/chainhash"
 	"github.com/btcsuite/btcd/wire"
 	"github.com/lightninglabs/taproot-assets/asset"
-	"github.com/lightninglabs/taproot-assets/chanutils"
 	"github.com/lightninglabs/taproot-assets/commitment"
+	"github.com/lightninglabs/taproot-assets/fn"
 	"github.com/lightninglabs/taproot-assets/mssmt"
 	"github.com/lightninglabs/taproot-assets/proof"
 	"github.com/lightninglabs/taproot-assets/tapgarden"
@@ -335,5 +335,5 @@ type Porter interface {
 
 	// EventPublisher is a subscription interface that allows callers to
 	// subscribe to events that are relevant to the Porter.
-	chanutils.EventPublisher[chanutils.Event, bool]
+	fn.EventPublisher[fn.Event, bool]
 }
diff --git a/tapfreighter/parcel.go b/tapfreighter/parcel.go
index e9937600..1b768068 100644
--- a/tapfreighter/parcel.go
+++ b/tapfreighter/parcel.go
@@ -10,8 +10,8 @@ import (
 	"github.com/btcsuite/btcd/wire"
 	"github.com/lightninglabs/taproot-assets/address"
 	"github.com/lightninglabs/taproot-assets/asset"
-	"github.com/lightninglabs/taproot-assets/chanutils"
 	"github.com/lightninglabs/taproot-assets/commitment"
+	"github.com/lightninglabs/taproot-assets/fn"
 	"github.com/lightninglabs/taproot-assets/mssmt"
 	"github.com/lightninglabs/taproot-assets/proof"
 	"github.com/lightninglabs/taproot-assets/tappsbt"
@@ -624,7 +624,7 @@ func addOtherOutputExclusionProofs(outputs []*tappsbt.VOutput,
 		log.Tracef("Generated exclusion proof for anchor output index "+
 			"%d with asset_id=%v, taproot_asset_root=%x, "+
 			"internal_key=%x", outIndex, asset.ID(),
-			chanutils.ByteSlice(tapTree.TapscriptRoot(nil)),
+			fn.ByteSlice(tapTree.TapscriptRoot(nil)),
 			vOut.AnchorOutputInternalKey.SerializeCompressed())
 
 		siblingPreimage := vOut.AnchorOutputTapscriptSibling
diff --git a/tapfreighter/wallet.go b/tapfreighter/wallet.go
index 69e303b8..4ef45c93 100644
--- a/tapfreighter/wallet.go
+++ b/tapfreighter/wallet.go
@@ -17,8 +17,8 @@ import (
 	"github.com/davecgh/go-spew/spew"
 	"github.com/lightninglabs/taproot-assets/address"
 	"github.com/lightninglabs/taproot-assets/asset"
-	"github.com/lightninglabs/taproot-assets/chanutils"
 	"github.com/lightninglabs/taproot-assets/commitment"
+	"github.com/lightninglabs/taproot-assets/fn"
 	"github.com/lightninglabs/taproot-assets/mssmt"
 	"github.com/lightninglabs/taproot-assets/proof"
 	"github.com/lightninglabs/taproot-assets/tapgarden"
@@ -427,7 +427,7 @@ func (f *AssetWallet) FundPacket(ctx context.Context,
 		)
 	}
 
-	inputsScriptKeys := chanutils.Map(
+	inputsScriptKeys := fn.Map(
 		vPkt.Inputs, func(vInput *tappsbt.VInput) *btcec.PublicKey {
 			return vInput.Asset().ScriptKey.PubKey
 		},
@@ -622,9 +622,8 @@ func (f *AssetWallet) setVPacketInputs(ctx context.Context,
 
 		log.Tracef("Input commitment taproot_asset_root=%x, "+
 			"internal_key=%x, pk_script=%x, trimmed_merkle_root=%x",
-			chanutils.ByteSlice(
-				assetInput.Commitment.TapscriptRoot(nil),
-			), internalKey.PubKey.SerializeCompressed(),
+			fn.ByteSlice(assetInput.Commitment.TapscriptRoot(nil)),
+			internalKey.PubKey.SerializeCompressed(),
 			anchorPkScript, anchorMerkleRoot[:])
 
 		// We'll also include an inclusion proof for the input asset in
diff --git a/tapgarden/caretaker.go b/tapgarden/caretaker.go
index 293d830f..194b9ca1 100644
--- a/tapgarden/caretaker.go
+++ b/tapgarden/caretaker.go
@@ -13,8 +13,8 @@ import (
 	"github.com/btcsuite/btcd/wire"
 	"github.com/davecgh/go-spew/spew"
 	"github.com/lightninglabs/taproot-assets/asset"
-	"github.com/lightninglabs/taproot-assets/chanutils"
 	"github.com/lightninglabs/taproot-assets/commitment"
+	"github.com/lightninglabs/taproot-assets/fn"
 	"github.com/lightninglabs/taproot-assets/proof"
 	"github.com/lightninglabs/taproot-assets/universe"
 	"github.com/lightningnetwork/lnd/chainntnfs"
@@ -99,7 +99,7 @@ type BatchCaretaker struct {
 
 	// ContextGuard provides a wait group and main quit channel that can be
 	// used to create guarded contexts.
-	*chanutils.ContextGuard
+	*fn.ContextGuard
 }
 
 // NewBatchCaretaker creates a new Taproot Asset caretaker based on the passed
@@ -111,7 +111,7 @@ func NewBatchCaretaker(cfg *BatchCaretakerConfig) *BatchCaretaker {
 		batchKey:  asset.ToSerialized(cfg.Batch.BatchKey.PubKey),
 		cfg:       cfg,
 		confEvent: make(chan *chainntnfs.TxConfirmation, 1),
-		ContextGuard: &chanutils.ContextGuard{
+		ContextGuard: &fn.ContextGuard{
 			DefaultTimeout: DefaultTimeout,
 			Quit:           make(chan struct{}),
 		},
diff --git a/tapgarden/custodian.go b/tapgarden/custodian.go
index 51ccbfcf..c9c35503 100644
--- a/tapgarden/custodian.go
+++ b/tapgarden/custodian.go
@@ -13,7 +13,7 @@ import (
 	"github.com/lightninglabs/lndclient"
 	"github.com/lightninglabs/taproot-assets/address"
 	"github.com/lightninglabs/taproot-assets/asset"
-	"github.com/lightninglabs/taproot-assets/chanutils"
+	"github.com/lightninglabs/taproot-assets/fn"
 	"github.com/lightninglabs/taproot-assets/proof"
 	"github.com/lightningnetwork/lnd/lnrpc"
 )
@@ -68,11 +68,11 @@ type Custodian struct {
 	// addrSubscription is the subscription queue through which we receive
 	// events about new addresses being created (and we also receive all
 	// previously existing addresses on startup).
-	addrSubscription *chanutils.EventReceiver[*address.AddrWithKeyInfo]
+	addrSubscription *fn.EventReceiver[*address.AddrWithKeyInfo]
 
 	// proofSubscription is the subscription queue through which we receive
 	// events about new proofs being imported.
-	proofSubscription *chanutils.EventReceiver[proof.Blob]
+	proofSubscription *fn.EventReceiver[proof.Blob]
 
 	// events is a map of all transaction outpoints and their ongoing
 	// address events of inbound assets.
@@ -80,24 +80,22 @@ type Custodian struct {
 
 	// ContextGuard provides a wait group and main quit channel that can be
 	// used to create guarded contexts.
-	*chanutils.ContextGuard
+	*fn.ContextGuard
 }
 
 // NewCustodian creates a new Taproot Asset custodian based on the passed
 // config.
 func NewCustodian(cfg *CustodianConfig) *Custodian {
-	addrSub := chanutils.NewEventReceiver[*address.AddrWithKeyInfo](
-		chanutils.DefaultQueueSize,
-	)
-	proofSub := chanutils.NewEventReceiver[proof.Blob](
-		chanutils.DefaultQueueSize,
+	addrSub := fn.NewEventReceiver[*address.AddrWithKeyInfo](
+		fn.DefaultQueueSize,
 	)
+	proofSub := fn.NewEventReceiver[proof.Blob](fn.DefaultQueueSize)
 	return &Custodian{
 		cfg:               cfg,
 		addrSubscription:  addrSub,
 		proofSubscription: proofSub,
 		events:            make(map[wire.OutPoint]*address.Event),
-		ContextGuard: &chanutils.ContextGuard{
+		ContextGuard: &fn.ContextGuard{
 			DefaultTimeout: DefaultTimeout,
 			Quit:           make(chan struct{}),
 		},
@@ -268,7 +266,7 @@ func (c *Custodian) watchInboundAssets() {
 		if err != nil {
 			// We'll report the error to the main daemon, but only
 			// if this isn't a context cancel.
-			if chanutils.IsCanceled(err) {
+			if fn.IsCanceled(err) {
 				return
 			}
 
@@ -498,7 +496,7 @@ func (c *Custodian) checkProofAvailable(event *address.Event) error {
 	// the proof is available in the relational database. If the proof is
 	// not in the DB, we can't update the event.
 	blob, err := c.cfg.ProofNotifier.FetchProof(ctxt, proof.Locator{
-		AssetID:   chanutils.Ptr(event.Addr.AssetID),
+		AssetID:   fn.Ptr(event.Addr.AssetID),
 		GroupKey:  event.Addr.GroupKey,
 		ScriptKey: event.Addr.ScriptKey,
 	})
diff --git a/tapgarden/planter.go b/tapgarden/planter.go
index 728241b6..a4ba0725 100644
--- a/tapgarden/planter.go
+++ b/tapgarden/planter.go
@@ -8,7 +8,7 @@ import (
 
 	"github.com/btcsuite/btcd/btcec/v2"
 	"github.com/lightninglabs/taproot-assets/asset"
-	"github.com/lightninglabs/taproot-assets/chanutils"
+	"github.com/lightninglabs/taproot-assets/fn"
 	"github.com/lightninglabs/taproot-assets/proof"
 	"github.com/lightninglabs/taproot-assets/universe"
 	"github.com/lightningnetwork/lnd/ticker"
@@ -185,7 +185,7 @@ type ChainPlanter struct {
 
 	// ContextGuard provides a wait group and main quit channel that can be
 	// used to create guarded contexts.
-	*chanutils.ContextGuard
+	*fn.ContextGuard
 }
 
 // NewChainPlanter creates a new ChainPlanter instance given the passed config.
@@ -196,7 +196,7 @@ func NewChainPlanter(cfg PlanterConfig) *ChainPlanter {
 		completionSignals: make(chan BatchKey),
 		seedlingReqs:      make(chan *Seedling),
 		stateReqs:         make(chan stateRequest),
-		ContextGuard: &chanutils.ContextGuard{
+		ContextGuard: &fn.ContextGuard{
 			DefaultTimeout: DefaultTimeout,
 			Quit:           make(chan struct{}),
 		},
@@ -618,7 +618,7 @@ func (c *ChainPlanter) gardener() {
 func (c *ChainPlanter) PendingBatch() (*MintingBatch, error) {
 	req := newStateReq[*MintingBatch](reqTypePendingBatch)
 
-	if !chanutils.SendOrQuit[stateRequest](c.stateReqs, req, c.Quit) {
+	if !fn.SendOrQuit[stateRequest](c.stateReqs, req, c.Quit) {
 		return nil, fmt.Errorf("chain planter shutting down")
 	}
 
@@ -630,7 +630,7 @@ func (c *ChainPlanter) PendingBatch() (*MintingBatch, error) {
 func (c *ChainPlanter) NumActiveBatches() (int, error) {
 	req := newStateReq[int](reqTypeNumActiveBatches)
 
-	if !chanutils.SendOrQuit[stateRequest](c.stateReqs, req, c.Quit) {
+	if !fn.SendOrQuit[stateRequest](c.stateReqs, req, c.Quit) {
 		return 0, fmt.Errorf("chain planter shutting down")
 	}
 
@@ -644,7 +644,7 @@ func (c *ChainPlanter) ListBatches(batchKey *btcec.PublicKey) ([]*MintingBatch,
 
 	req := newStateParamReq[[]*MintingBatch](reqTypeListBatches, batchKey)
 
-	if !chanutils.SendOrQuit[stateRequest](c.stateReqs, req, c.Quit) {
+	if !fn.SendOrQuit[stateRequest](c.stateReqs, req, c.Quit) {
 		return nil, fmt.Errorf("chain planter shutting down")
 	}
 
@@ -655,7 +655,7 @@ func (c *ChainPlanter) ListBatches(batchKey *btcec.PublicKey) ([]*MintingBatch,
 func (c *ChainPlanter) FinalizeBatch() (*btcec.PublicKey, error) {
 	req := newStateReq[*btcec.PublicKey](reqTypeFinalizeBatch)
 
-	if !chanutils.SendOrQuit[stateRequest](c.stateReqs, req, c.Quit) {
+	if !fn.SendOrQuit[stateRequest](c.stateReqs, req, c.Quit) {
 		return nil, fmt.Errorf("chain planter shutting down")
 	}
 
@@ -666,7 +666,7 @@ func (c *ChainPlanter) FinalizeBatch() (*btcec.PublicKey, error) {
 func (c *ChainPlanter) CancelBatch() (*btcec.PublicKey, error) {
 	req := newStateReq[*btcec.PublicKey](reqTypeCancelBatch)
 
-	if !chanutils.SendOrQuit[stateRequest](c.stateReqs, req, c.Quit) {
+	if !fn.SendOrQuit[stateRequest](c.stateReqs, req, c.Quit) {
 		return nil, fmt.Errorf("chain planter shutting down")
 	}
 
@@ -809,7 +809,7 @@ func (c *ChainPlanter) QueueNewSeedling(req *Seedling) (SeedlingUpdates, error)
 
 	// Attempt to send the new request, or exit if the quit channel
 	// triggered first.
-	if !chanutils.SendOrQuit(c.seedlingReqs, req, c.Quit) {
+	if !fn.SendOrQuit(c.seedlingReqs, req, c.Quit) {
 		return nil, fmt.Errorf("planter shutting down")
 	}
 
diff --git a/tappsbt/encode.go b/tappsbt/encode.go
index f2fe9160..b9697082 100644
--- a/tappsbt/encode.go
+++ b/tappsbt/encode.go
@@ -13,8 +13,8 @@ import (
 	"github.com/btcsuite/btcd/txscript"
 	"github.com/btcsuite/btcd/wire"
 	"github.com/lightninglabs/taproot-assets/asset"
-	"github.com/lightninglabs/taproot-assets/chanutils"
 	"github.com/lightninglabs/taproot-assets/commitment"
+	"github.com/lightninglabs/taproot-assets/fn"
 	"github.com/lightningnetwork/lnd/tlv"
 )
 
@@ -280,7 +280,7 @@ func tlvEncoder(val any, enc tlv.Encoder) encoderFunc {
 
 		return []*customPsbtField{
 			{
-				Key:   chanutils.CopySlice(key),
+				Key:   fn.CopySlice(key),
 				Value: b.Bytes(),
 			},
 		}, nil
@@ -314,11 +314,11 @@ func assetEncoder(a *asset.Asset) encoderFunc {
 func booleanEncoder(val bool) encoderFunc {
 	return func(key []byte) ([]*customPsbtField, error) {
 		unknown := &customPsbtField{
-			Key:   chanutils.CopySlice(key),
-			Value: chanutils.CopySlice(falseAsBytes),
+			Key:   fn.CopySlice(key),
+			Value: fn.CopySlice(falseAsBytes),
 		}
 		if val {
-			unknown.Value = chanutils.CopySlice(trueAsBytes)
+			unknown.Value = fn.CopySlice(trueAsBytes)
 		}
 
 		return []*customPsbtField{unknown}, nil
@@ -337,7 +337,7 @@ func bip32DerivationEncoder(derivations []*psbt.Bip32Derivation) encoderFunc {
 		for idx := range derivations {
 			d := derivations[idx]
 
-			keyCopy := chanutils.CopySlice(key)
+			keyCopy := fn.CopySlice(key)
 			unknowns[idx] = &customPsbtField{
 				Key: append(keyCopy, d.PubKey...),
 				Value: psbt.SerializeBIP32Derivation(
@@ -368,7 +368,7 @@ func taprootBip32DerivationEncoder(
 				return nil, err
 			}
 
-			keyCopy := chanutils.CopySlice(key)
+			keyCopy := fn.CopySlice(key)
 			unknowns[idx] = &customPsbtField{
 				Key:   append(keyCopy, d.XOnlyPubKey...),
 				Value: value,
diff --git a/tappsbt/interface.go b/tappsbt/interface.go
index cd5c2d71..a773122c 100644
--- a/tappsbt/interface.go
+++ b/tappsbt/interface.go
@@ -11,8 +11,8 @@ import (
 	"github.com/btcsuite/btcd/txscript"
 	"github.com/lightninglabs/taproot-assets/address"
 	"github.com/lightninglabs/taproot-assets/asset"
-	"github.com/lightninglabs/taproot-assets/chanutils"
 	"github.com/lightninglabs/taproot-assets/commitment"
+	"github.com/lightninglabs/taproot-assets/fn"
 	"github.com/lightningnetwork/lnd/keychain"
 )
 
@@ -196,31 +196,31 @@ func (p *VPacket) HasSplitCommitment() (bool, error) {
 // HasSplitRootOutput determines if this virtual transaction has a split root
 // output.
 func (p *VPacket) HasSplitRootOutput() bool {
-	return chanutils.Any(p.Outputs, VOutIsSplitRoot)
+	return fn.Any(p.Outputs, VOutIsSplitRoot)
 }
 
 // HasInteractiveOutput determines if this virtual transaction has an
 // interactive output.
 func (p *VPacket) HasInteractiveOutput() bool {
-	return chanutils.Any(p.Outputs, VOutIsInteractive)
+	return fn.Any(p.Outputs, VOutIsInteractive)
 }
 
 // SplitRootOutput returns the split root output in the virtual transaction, or
 // an error if there is none or more than one.
 func (p *VPacket) SplitRootOutput() (*VOutput, error) {
-	count := chanutils.Count(p.Outputs, VOutIsSplitRoot)
+	count := fn.Count(p.Outputs, VOutIsSplitRoot)
 	if count != 1 {
 		return nil, fmt.Errorf("expected 1 split root output, got %d",
 			count)
 	}
 
-	return chanutils.First(p.Outputs, VOutIsSplitRoot)
+	return fn.First(p.Outputs, VOutIsSplitRoot)
 }
 
 // FirstNonSplitRootOutput returns the first non-change output in the virtual
 // transaction.
 func (p *VPacket) FirstNonSplitRootOutput() (*VOutput, error) {
-	result, err := chanutils.First(p.Outputs, VOutIsNotSplitRoot)
+	result, err := fn.First(p.Outputs, VOutIsNotSplitRoot)
 	if err != nil {
 		return nil, fmt.Errorf("no non split root output found")
 	}
@@ -231,7 +231,7 @@ func (p *VPacket) FirstNonSplitRootOutput() (*VOutput, error) {
 // FirstInteractiveOutput returns the first interactive output in the virtual
 // transaction.
 func (p *VPacket) FirstInteractiveOutput() (*VOutput, error) {
-	result, err := chanutils.First(p.Outputs, VOutIsInteractive)
+	result, err := fn.First(p.Outputs, VOutIsInteractive)
 	if err != nil {
 		return nil, fmt.Errorf("no interactive output found")
 	}
@@ -602,7 +602,7 @@ func AddBip32Derivation(derivations []*psbt.Bip32Derivation,
 	}
 
 	predicate := bip32DerivationKeyEqual(target.PubKey)
-	if chanutils.Any(derivations, predicate) {
+	if fn.Any(derivations, predicate) {
 		return derivations
 	}
 
@@ -619,7 +619,7 @@ func AddTaprootBip32Derivation(derivations []*psbt.TaprootBip32Derivation,
 	}
 
 	predicate := taprootBip32DerivationKeyEqual(target.XOnlyPubKey)
-	if chanutils.Any(derivations, predicate) {
+	if fn.Any(derivations, predicate) {
 		return derivations
 	}
 
diff --git a/tapscript/send.go b/tapscript/send.go
index cc27fd1b..0273434f 100644
--- a/tapscript/send.go
+++ b/tapscript/send.go
@@ -724,7 +724,7 @@ func CreateOutputCommitments(inputTapCommitments tappsbt.InputCommitments,
 
 	// Merge all input Taproot Asset commitments into a single commitment.
 	//
-	// TODO(ffranr): Use `chanutils.ForEach` and `inputTapCommitments[1:]`.
+	// TODO(ffranr): Use `fn.ForEach` and `inputTapCommitments[1:]`.
 	inputTapCommitment := inputTapCommitments[0]
 	for idx := range inputTapCommitments {
 		if idx == 0 {
diff --git a/universe/auto_syncer.go b/universe/auto_syncer.go
index 22a81fa9..1ea3a14b 100644
--- a/universe/auto_syncer.go
+++ b/universe/auto_syncer.go
@@ -7,7 +7,7 @@ import (
 	"time"
 
 	"github.com/davecgh/go-spew/spew"
-	"github.com/lightninglabs/taproot-assets/chanutils"
+	"github.com/lightninglabs/taproot-assets/fn"
 )
 
 const (
@@ -75,7 +75,7 @@ type FederationPushReq struct {
 type FederationEnvoy struct {
 	cfg FederationConfig
 
-	*chanutils.ContextGuard
+	*fn.ContextGuard
 
 	startOnce sync.Once
 
@@ -89,7 +89,7 @@ func NewFederationEnvoy(cfg FederationConfig) *FederationEnvoy {
 	return &FederationEnvoy{
 		cfg:          cfg,
 		pushRequests: make(chan *FederationPushReq),
-		ContextGuard: &chanutils.ContextGuard{
+		ContextGuard: &fn.ContextGuard{
 			DefaultTimeout: DefaultTimeout,
 			Quit:           make(chan struct{}),
 		},
@@ -104,7 +104,7 @@ func (f *FederationEnvoy) Start() error {
 		// Before we start the main goroutine, we'll add the set of
 		// static Universe servers.
 		addrs := f.cfg.StaticFederationMembers
-		serverAddrs := chanutils.Map(addrs, func(a string) ServerAddr {
+		serverAddrs := fn.Map(addrs, func(a string) ServerAddr {
 			return NewServerAddrFromStr(a)
 		})
 		err := f.AddServer(serverAddrs...)
@@ -234,7 +234,7 @@ func (f *FederationEnvoy) pushProofToFederation(uniID Identifier,
 
 	// To conclude, we'll attempt to push the new proof to all the universe
 	// servers in parallel.
-	err = chanutils.ParSlice(ctx, fedServers, pushNewProof)
+	err = fn.ParSlice(ctx, fedServers, pushNewProof)
 	if err != nil {
 		// TODO(roasbeef): retry in the background until successful?
 		log.Errorf("unable to push proof to federation: %v", err)
@@ -284,7 +284,7 @@ func (f *FederationEnvoy) syncer() {
 
 			// Sync the set of servers in parallel, waiting until
 			// the syncs are finished to proceed.
-			err = chanutils.ParSlice(
+			err = fn.ParSlice(
 				ctx, fedServers, f.syncUniverseState,
 			)
 			if err != nil {
@@ -352,11 +352,11 @@ func (f *FederationEnvoy) RegisterIssuance(ctx context.Context, id Identifier,
 		err:  make(chan error, 1),
 	}
 
-	if !chanutils.SendOrQuit(f.pushRequests, pushReq, f.Quit) {
+	if !fn.SendOrQuit(f.pushRequests, pushReq, f.Quit) {
 		return nil, fmt.Errorf("unable to push new proof event")
 	}
 
-	return chanutils.RecvResp(pushReq.resp, pushReq.err, f.Quit)
+	return fn.RecvResp(pushReq.resp, pushReq.err, f.Quit)
 }
 
 // AddServers adds a new set of servers to the federation, then immediately
@@ -379,7 +379,7 @@ func (f *FederationEnvoy) AddServer(addrs ...ServerAddr) error {
 		ctx, cancel = f.WithCtxQuitNoTimeout()
 		defer cancel()
 
-		err := chanutils.ParSlice(ctx, addrs, f.syncUniverseState)
+		err := fn.ParSlice(ctx, addrs, f.syncUniverseState)
 		if err != nil {
 			log.Warnf("unable to sync universe state: %v", err)
 		}
diff --git a/universe/base.go b/universe/base.go
index 66335839..afc5edcb 100644
--- a/universe/base.go
+++ b/universe/base.go
@@ -217,7 +217,8 @@ func (a *MintingArchive) RegisterIssuance(ctx context.Context, id Identifier,
 			context.Background(), id, key,
 		)
 		if err != nil {
-			log.Warnf("unable to log new proof event: %v", err)
+			log.Warnf("unable to log new proof event (id=%v): %v",
+				spew.Sdump(id), err)
 		}
 	}()
 
@@ -240,7 +241,9 @@ func (a *MintingArchive) FetchIssuanceProof(ctx context.Context, id Identifier,
 				context.Background(), id, key,
 			)
 			if err != nil {
-				log.Warnf("unable to log sync event: %v", err)
+				log.Warnf("unable to log sync event "+
+					"(id=%v, key=%v): %v", spew.Sdump(id),
+					spew.Sdump(key), err)
 			}
 		}()
 	}()
diff --git a/universe/syncer.go b/universe/syncer.go
index 71624195..d3b2780b 100644
--- a/universe/syncer.go
+++ b/universe/syncer.go
@@ -6,7 +6,7 @@ import (
 	"fmt"
 
 	"github.com/davecgh/go-spew/spew"
-	"github.com/lightninglabs/taproot-assets/chanutils"
+	"github.com/lightninglabs/taproot-assets/fn"
 	"github.com/lightninglabs/taproot-assets/mssmt"
 )
 
@@ -68,8 +68,7 @@ func (s *SimpleSyncer) executeSync(ctx context.Context, diffEngine DiffEngine,
 		// as a series of parallel requests backed by a worker pool.
 		//
 		// TODO(roasbeef): can actually make non-blocking..
-		err = chanutils.ParSlice(
-			ctx, idsToSync,
+		err = fn.ParSlice(ctx, idsToSync,
 			func(ctx context.Context, id Identifier) error {
 				root, err := diffEngine.RootNode(ctx, id)
 				if err != nil {
@@ -92,14 +91,14 @@ func (s *SimpleSyncer) executeSync(ctx context.Context, diffEngine DiffEngine,
 		rootsToSync = make(chan BaseRoot, len(roots))
 
 		// TODO(roasbeef): can make non-blocking w/ goroutine
-		chanutils.SendAll(rootsToSync, roots...)
+		fn.SendAll(rootsToSync, roots...)
 	}
 	if err != nil {
 		return nil, fmt.Errorf("unable to fetch roots for "+
 			"universe sync: %w", err)
 	}
 
-	targetRoots := chanutils.Collect(rootsToSync)
+	targetRoots := fn.Collect(rootsToSync)
 
 	log.Infof("Obtained %v roots from remote Universe server",
 		len(targetRoots))
@@ -109,7 +108,7 @@ func (s *SimpleSyncer) executeSync(ctx context.Context, diffEngine DiffEngine,
 	// Now that we know the set of Universes we need to sync, we'll execute
 	// the diff operation for each of them.
 	syncDiffs := make(chan AssetSyncDiff, len(targetRoots))
-	err = chanutils.ParSlice(ctx, targetRoots, func(ctx context.Context, remoteRoot BaseRoot) error {
+	err = fn.ParSlice(ctx, targetRoots, func(ctx context.Context, remoteRoot BaseRoot) error {
 		// First, we'll compare the remote root against the local root.
 		uniID := remoteRoot.ID
 		localRoot, err := s.cfg.LocalDiffEngine.RootNode(ctx, uniID)
@@ -150,7 +149,7 @@ func (s *SimpleSyncer) executeSync(ctx context.Context, diffEngine DiffEngine,
 
 		// With the set of keys fetched, we can now find the set of
 		// keys that need to be synced.
-		keysToFetch := chanutils.SetDiff(remoteUnikeys, localUnikeys)
+		keysToFetch := fn.SetDiff(remoteUnikeys, localUnikeys)
 
 		log.Infof("UniverseRoot(%v): diff_size=%v", uniID.String(),
 			len(keysToFetch))
@@ -161,7 +160,7 @@ func (s *SimpleSyncer) executeSync(ctx context.Context, diffEngine DiffEngine,
 		// Now that we know where the divergence is, we can fetch the
 		// issuance proofs from the remote party.
 		newLeaves := make(chan *MintingLeaf, len(keysToFetch))
-		err = chanutils.ParSlice(ctx, keysToFetch, func(ctx context.Context, key BaseKey) error {
+		err = fn.ParSlice(ctx, keysToFetch, func(ctx context.Context, key BaseKey) error {
 			newProof, err := diffEngine.FetchIssuanceProof(ctx, uniID, key)
 			if err != nil {
 				return err
@@ -213,7 +212,7 @@ func (s *SimpleSyncer) executeSync(ctx context.Context, diffEngine DiffEngine,
 		syncDiffs <- AssetSyncDiff{
 			OldUniverseRoot: localRoot,
 			NewUniverseRoot: remoteRoot,
-			NewLeafProofs:   chanutils.Collect(newLeaves),
+			NewLeafProofs:   fn.Collect(newLeaves),
 		}
 
 		log.Infof("Sync for UniverseRoot(%v) complete!", uniID.String())
@@ -228,7 +227,7 @@ func (s *SimpleSyncer) executeSync(ctx context.Context, diffEngine DiffEngine,
 	}
 
 	// Finally, we'll collect all the diffs and return them to the caller.
-	return chanutils.Collect(syncDiffs), nil
+	return fn.Collect(syncDiffs), nil
 }
 
 // SyncUniverse attempts to synchronize the local universe with the remote
diff --git a/universe_rpc_registrar.go b/universe_rpc_registrar.go
index cf488ad2..9b75e0fd 100644
--- a/universe_rpc_registrar.go
+++ b/universe_rpc_registrar.go
@@ -6,7 +6,7 @@ import (
 	"crypto/tls"
 	"fmt"
 
-	"github.com/lightninglabs/taproot-assets/chanutils"
+	"github.com/lightninglabs/taproot-assets/fn"
 	"github.com/lightninglabs/taproot-assets/mssmt"
 	unirpc "github.com/lightninglabs/taproot-assets/taprpc/universerpc"
 	"github.com/lightninglabs/taproot-assets/universe"
@@ -84,7 +84,7 @@ func unmarshalIssuanceProof(ctx context.Context,
 	return &universe.IssuanceProof{
 		MintingKey: baseKey,
 		UniverseRoot: mssmt.NewComputedBranch(
-			chanutils.ToArray[mssmt.NodeHash](
+			fn.ToArray[mssmt.NodeHash](
 				proofResp.UniverseRoot.MssmtRoot.RootHash,
 			),
 			uint64(proofResp.UniverseRoot.MssmtRoot.RootSum),


Test patch:
diff --git a/chanutils/concurrency_test.go b/fn/concurrency_test.go
similarity index 98%
rename from chanutils/concurrency_test.go
rename to fn/concurrency_test.go
index 31bbebe2..1cfe2acf 100644
--- a/chanutils/concurrency_test.go
+++ b/fn/concurrency_test.go
@@ -1,4 +1,4 @@
-package chanutils
+package fn
 
 import (
 	"context"
diff --git a/chanutils/errors_test.go b/fn/errors_test.go
similarity index 96%
rename from chanutils/errors_test.go
rename to fn/errors_test.go
index bcc14235..20884db5 100644
--- a/chanutils/errors_test.go
+++ b/fn/errors_test.go
@@ -1,4 +1,4 @@
-package chanutils
+package fn
 
 import (
 	"context"
diff --git a/itest/assertions.go b/itest/assertions.go
index 5cbbadb4..72b06c01 100644
--- a/itest/assertions.go
+++ b/itest/assertions.go
@@ -12,7 +12,7 @@ import (
 	"github.com/btcsuite/btcd/txscript"
 	"github.com/btcsuite/btcd/wire"
 	"github.com/lightninglabs/taproot-assets/asset"
-	"github.com/lightninglabs/taproot-assets/chanutils"
+	"github.com/lightninglabs/taproot-assets/fn"
 	"github.com/lightninglabs/taproot-assets/proof"
 	"github.com/lightninglabs/taproot-assets/taprpc"
 	unirpc "github.com/lightninglabs/taproot-assets/taprpc/universerpc"
@@ -688,7 +688,7 @@ func assertUniverseRootsEqual(t *testing.T, a, b *unirpc.AssetRootResponse) {
 	// same set of asset IDs are being tracked.
 	uniKeys := maps.Keys(a.UniverseRoots)
 	require.Equal(t, len(a.UniverseRoots), len(b.UniverseRoots))
-	require.True(t, chanutils.All(uniKeys, func(key string) bool {
+	require.True(t, fn.All(uniKeys, func(key string) bool {
 		_, ok := b.UniverseRoots[key]
 		return ok
 	}))
@@ -805,7 +805,7 @@ func assertUniverseAssetStats(t *testing.T, node *tapdHarness,
 	require.Len(t, assetStats.AssetStats, len(assetIDs))
 
 	for _, assetStat := range assetStats.AssetStats {
-		found := chanutils.Any(assetIDs, func(id []byte) bool {
+		found := fn.Any(assetIDs, func(id []byte) bool {
 			return bytes.Equal(assetStat.AssetId, id)
 		})
 		require.True(t, found)
diff --git a/itest/assets_test.go b/itest/assets_test.go
index 633cea53..6f0d4e55 100644
--- a/itest/assets_test.go
+++ b/itest/assets_test.go
@@ -9,7 +9,7 @@ import (
 	"time"
 
 	"github.com/btcsuite/btcd/chaincfg/chainhash"
-	"github.com/lightninglabs/taproot-assets/chanutils"
+	"github.com/lightninglabs/taproot-assets/fn"
 	"github.com/lightninglabs/taproot-assets/proof"
 	"github.com/lightninglabs/taproot-assets/taprpc"
 	"github.com/lightninglabs/taproot-assets/taprpc/mintrpc"
@@ -469,7 +469,7 @@ func testMintAssetNameCollisionError(t *harnessTest) {
 
 		return batch.Assets[0].AssetType == taprpc.AssetType_COLLECTIBLE
 	}
-	batchCollide, err := chanutils.First(allBatches, isCollidingBatch)
+	batchCollide, err := fn.First(allBatches, isCollidingBatch)
 	require.NoError(t.t, err)
 
 	require.Len(t.t, batchCollide.Assets, 1)
diff --git a/itest/multi_asset_group_test.go b/itest/multi_asset_group_test.go
index 96dc2520..df9b60e1 100644
--- a/itest/multi_asset_group_test.go
+++ b/itest/multi_asset_group_test.go
@@ -7,7 +7,7 @@ import (
 	"testing"
 
 	"github.com/lightninglabs/taproot-assets/asset"
-	"github.com/lightninglabs/taproot-assets/chanutils"
+	"github.com/lightninglabs/taproot-assets/fn"
 	"github.com/lightninglabs/taproot-assets/taprpc"
 	"github.com/lightninglabs/taproot-assets/taprpc/mintrpc"
 	"github.com/stretchr/testify/require"
@@ -91,7 +91,7 @@ func testMintMultiAssetGroups(t *harnessTest) {
 
 	// We need to fetch the minted group anchor asset, which includes the
 	// genesis information used to compute the tweak for the group key.
-	normalAnchor, err := chanutils.First(
+	normalAnchor, err := fn.First(
 		mintedBatch, func(asset *taprpc.Asset) bool {
 			return matchingName(asset, issuableAssets[0].Asset.Name)
 		},
@@ -104,7 +104,7 @@ func testMintMultiAssetGroups(t *harnessTest) {
 		normalAnchor.AssetGroup.TweakedGroupKey,
 	)
 
-	collectAnchor, err := chanutils.First(
+	collectAnchor, err := fn.First(
 		mintedBatch, func(asset *taprpc.Asset) bool {
 			return matchingName(asset, issuableAssets[1].Asset.Name)
 		},
@@ -130,7 +130,7 @@ func testMintMultiAssetGroups(t *harnessTest) {
 		require.NoError(t.t, secondTapd.stop(true))
 	}()
 
-	normalMember, err := chanutils.First(
+	normalMember, err := fn.First(
 		mintedBatch, func(asset *taprpc.Asset) bool {
 			return asset.Amount == normalAnchor.Amount/2
 		},
@@ -175,7 +175,7 @@ func testMintMultiAssetGroups(t *harnessTest) {
 		)
 		return isNotAnchor && isGrouped
 	}
-	collectMember, err := chanutils.First(mintedBatch, isCollectGroupMember)
+	collectMember, err := fn.First(mintedBatch, isCollectGroupMember)
 	require.NoError(t.t, err)
 
 	collectMemberGenInfo := collectMember.AssetGenesis
diff --git a/itest/universe_test.go b/itest/universe_test.go
index 44b577f1..93a97562 100644
--- a/itest/universe_test.go
+++ b/itest/universe_test.go
@@ -8,7 +8,7 @@ import (
 	"io"
 
 	"github.com/btcsuite/btcd/btcec/v2/schnorr"
-	"github.com/lightninglabs/taproot-assets/chanutils"
+	"github.com/lightninglabs/taproot-assets/fn"
 	"github.com/lightninglabs/taproot-assets/taprpc"
 	unirpc "github.com/lightninglabs/taproot-assets/taprpc/universerpc"
 	"github.com/lightningnetwork/lnd/lntest/wait"
@@ -116,10 +116,9 @@ func testUniverseSync(t *harnessTest) {
 	// Finally, we'll ensure that the universe keys and leaves matches for
 	// both parties.
 	uniRoots := maps.Values(universeRoots.UniverseRoots)
-	uniIDs := chanutils.Map(uniRoots,
-		func(root *unirpc.UniverseRoot) *unirpc.ID {
-			return root.Id
-		},
+	uniIDs := fn.Map(uniRoots, func(root *unirpc.UniverseRoot) *unirpc.ID {
+		return root.Id
+	},
 	)
 	assertUniverseKeysEqual(t.t, uniIDs, t.tapd, bob)
 	assertUniverseLeavesEqual(t.t, uniIDs, t.tapd, bob)
diff --git a/tapdb/asset_minting_test.go b/tapdb/asset_minting_test.go
index 4651fb05..ce6a9284 100644
--- a/tapdb/asset_minting_test.go
+++ b/tapdb/asset_minting_test.go
@@ -15,8 +15,8 @@ import (
 	"github.com/btcsuite/btcd/chaincfg/chainhash"
 	"github.com/btcsuite/btcd/wire"
 	"github.com/lightninglabs/taproot-assets/asset"
-	"github.com/lightninglabs/taproot-assets/chanutils"
 	"github.com/lightninglabs/taproot-assets/commitment"
+	"github.com/lightninglabs/taproot-assets/fn"
 	"github.com/lightninglabs/taproot-assets/proof"
 	"github.com/lightninglabs/taproot-assets/tapdb/sqlc"
 	"github.com/lightninglabs/taproot-assets/tapgarden"
@@ -744,7 +744,7 @@ func TestCommitBatchChainActions(t *testing.T) {
 
 		return count
 	}
-	numKeyGroups := chanutils.Reduce(mintedAssets, keyGroupSumReducer)
+	numKeyGroups := fn.Reduce(mintedAssets, keyGroupSumReducer)
 	assetBalancesByGroup, err := confAssets.QueryAssetBalancesByGroup(
 		ctx, nil,
 	)
diff --git a/tapdb/universe_stats_test.go b/tapdb/universe_stats_test.go
index c4cfbac8..dfccf91a 100644
--- a/tapdb/universe_stats_test.go
+++ b/tapdb/universe_stats_test.go
@@ -9,7 +9,7 @@ import (
 	"testing"
 
 	"github.com/lightninglabs/taproot-assets/asset"
-	"github.com/lightninglabs/taproot-assets/chanutils"
+	"github.com/lightninglabs/taproot-assets/fn"
 	"github.com/lightninglabs/taproot-assets/tapdb/sqlc"
 	"github.com/lightninglabs/taproot-assets/universe"
 	"github.com/stretchr/testify/require"
@@ -323,10 +323,9 @@ func TestUniverseQuerySyncFilters(t *testing.T) {
 		},
 		{
 			name:       "type",
-			typeFilter: chanutils.Ptr(asset.Type(rand.Int() % 2)),
+			typeFilter: fn.Ptr(asset.Type(rand.Int() % 2)),
 			queryCheck: func(s *universe.AssetSyncStats) bool {
-				typeCount := chanutils.Reduce(
-					sh.universeLeaves,
+				typeCount := fn.Reduce(sh.universeLeaves,
 					func(acc int, p *universe.IssuanceProof) int {
 						if p.Leaf.Type == *s.Query.AssetTypeFilter {
 							return acc + 1
diff --git a/tapdb/universe_test.go b/tapdb/universe_test.go
index ec073a3c..9b91bc92 100644
--- a/tapdb/universe_test.go
+++ b/tapdb/universe_test.go
@@ -9,7 +9,7 @@ import (
 
 	"github.com/btcsuite/btcd/btcec/v2"
 	"github.com/lightninglabs/taproot-assets/asset"
-	"github.com/lightninglabs/taproot-assets/chanutils"
+	"github.com/lightninglabs/taproot-assets/fn"
 	"github.com/lightninglabs/taproot-assets/internal/test"
 	"github.com/lightninglabs/taproot-assets/mssmt"
 	"github.com/lightninglabs/taproot-assets/proof"
@@ -79,7 +79,7 @@ func TestUniverseEmptyTree(t *testing.T) {
 func randBaseKey(t *testing.T) universe.BaseKey {
 	return universe.BaseKey{
 		MintingOutpoint: test.RandOp(t),
-		ScriptKey: chanutils.Ptr(
+		ScriptKey: fn.Ptr(
 			asset.NewScriptKey(test.RandPubKey(t)),
 		),
 	}
@@ -206,7 +206,7 @@ func TestUniverseIssuanceProofs(t *testing.T) {
 	require.Equal(t, numLeaves, len(mintingKeys))
 
 	// The set of leaves we created above should match what was returned.
-	require.True(t, chanutils.All(mintingKeys, func(key universe.BaseKey) bool {
+	require.True(t, fn.All(mintingKeys, func(key universe.BaseKey) bool {
 		for _, testLeaf := range testLeaves {
 			if reflect.DeepEqual(key, testLeaf.BaseKey) {
 				return true
@@ -221,7 +221,7 @@ func TestUniverseIssuanceProofs(t *testing.T) {
 	dbLeaves, err := baseUniverse.MintingLeaves(ctx)
 	require.NoError(t, err)
 	require.Equal(t, numLeaves, len(dbLeaves))
-	require.True(t, chanutils.All(dbLeaves, func(leaf universe.MintingLeaf) bool {
+	require.True(t, fn.All(dbLeaves, func(leaf universe.MintingLeaf) bool {
 		for _, testLeaf := range testLeaves {
 			if leaf.Genesis.ID() == testLeaf.MintingLeaf.Genesis.ID() {
 				return true
@@ -343,7 +343,7 @@ func TestUniverseTreeIsolation(t *testing.T) {
 	require.NoError(t, err)
 
 	// We should be able to find both of the roots we've inserted above.
-	require.True(t, chanutils.All(rootNodes, func(rootNode universe.BaseRoot) bool {
+	require.True(t, fn.All(rootNodes, func(rootNode universe.BaseRoot) bool {
 		for _, rootNode := range rootNodes {
 			if mssmt.IsEqualNode(rootNode.Node, groupRoot) {
 				return true
diff --git a/tapgarden/custodian_test.go b/tapgarden/custodian_test.go
index 098d2ed7..8f3df739 100644
--- a/tapgarden/custodian_test.go
+++ b/tapgarden/custodian_test.go
@@ -13,7 +13,7 @@ import (
 	"github.com/lightninglabs/lndclient"
 	"github.com/lightninglabs/taproot-assets/address"
 	"github.com/lightninglabs/taproot-assets/asset"
-	"github.com/lightninglabs/taproot-assets/chanutils"
+	"github.com/lightninglabs/taproot-assets/fn"
 	"github.com/lightninglabs/taproot-assets/internal/test"
 	"github.com/lightninglabs/taproot-assets/proof"
 	"github.com/lightninglabs/taproot-assets/tapdb"
@@ -90,13 +90,13 @@ type custodianHarness struct {
 // assertStartup makes sure the custodian was started correctly.
 func (h *custodianHarness) assertStartup() {
 	// Make sure SubscribeTransactions is called on startup.
-	_, err := chanutils.RecvOrTimeout(
+	_, err := fn.RecvOrTimeout(
 		h.walletAnchor.SubscribeTxSignal, testTimeout,
 	)
 	require.NoError(h.t, err)
 
 	// Make sure ListTransactions is called on startup.
-	_, err = chanutils.RecvOrTimeout(
+	_, err = fn.RecvOrTimeout(
 		h.walletAnchor.ListTxnsSignal, testTimeout,
 	)
 	require.NoError(h.t, err)
@@ -114,7 +114,7 @@ func (h *custodianHarness) assertAddrsRegistered(
 	addrs ...*address.AddrWithKeyInfo) {
 
 	for _, addr := range addrs {
-		pubKey, err := chanutils.RecvOrTimeout(
+		pubKey, err := fn.RecvOrTimeout(
 			h.walletAnchor.ImportPubKeySignal, testTimeout,
 		)
 		require.NoError(h.t, err)
diff --git a/tapgarden/planter_test.go b/tapgarden/planter_test.go
index a0f4199a..9e4d9a7d 100644
--- a/tapgarden/planter_test.go
+++ b/tapgarden/planter_test.go
@@ -19,7 +19,7 @@ import (
 	"github.com/btcsuite/btcd/wire"
 	"github.com/davecgh/go-spew/spew"
 	"github.com/lightninglabs/taproot-assets/asset"
-	"github.com/lightninglabs/taproot-assets/chanutils"
+	"github.com/lightninglabs/taproot-assets/fn"
 	"github.com/lightninglabs/taproot-assets/internal/test"
 	"github.com/lightninglabs/taproot-assets/proof"
 	"github.com/lightninglabs/taproot-assets/tapdb"
@@ -154,7 +154,7 @@ func (t *mintingTestHarness) newRandSeedlings(numSeedlings int) []*tapgarden.See
 func (t *mintingTestHarness) assertKeyDerived() *keychain.KeyDescriptor {
 	t.Helper()
 
-	key, err := chanutils.RecvOrTimeout(t.keyRing.ReqKeys, defaultTimeout)
+	key, err := fn.RecvOrTimeout(t.keyRing.ReqKeys, defaultTimeout)
 	require.NoError(t, err)
 
 	return *key
@@ -181,7 +181,7 @@ func (t *mintingTestHarness) queueSeedlingsInBatch(
 
 		// We should get an update from the update channel that the
 		// seedling is now pending.
-		update, err := chanutils.RecvOrTimeout(updates, defaultTimeout)
+		update, err := fn.RecvOrTimeout(updates, defaultTimeout)
 		require.NoError(t, err)
 
 		// Make sure the seedling was planted without error.
@@ -273,12 +273,12 @@ func (t *mintingTestHarness) assertNumCaretakersActive(n int) {
 func (t *mintingTestHarness) assertGenesisTxFunded() *tapgarden.FundedPsbt {
 	// In order to fund a transaction, we expect a call to estimate the
 	// fee, followed by a request to fund a new PSBT packet.
-	_, err := chanutils.RecvOrTimeout(
+	_, err := fn.RecvOrTimeout(
 		t.chain.FeeEstimateSignal, defaultTimeout,
 	)
 	require.NoError(t, err)
 
-	pkt, err := chanutils.RecvOrTimeout(
+	pkt, err := fn.RecvOrTimeout(
 		t.wallet.FundPsbtSignal, defaultTimeout,
 	)
 	require.NoError(t, err)
@@ -331,7 +331,7 @@ func (t *mintingTestHarness) assertSeedlingsExist(
 		isNotCancelledBatch := func(batch *tapgarden.MintingBatch) bool {
 			return !isCancelledBatch(batch)
 		}
-		pendingBatch, err = chanutils.First(
+		pendingBatch, err = fn.First(
 			pendingBatches, isNotCancelledBatch,
 		)
 		require.NoError(t, err)
@@ -339,7 +339,7 @@ func (t *mintingTestHarness) assertSeedlingsExist(
 
 	if batchKey != nil {
 		var err error
-		pendingBatch, err = chanutils.First(pendingBatches, matchingBatch)
+		pendingBatch, err = fn.First(pendingBatches, matchingBatch)
 		require.NoError(t, err)
 	}
 
@@ -401,7 +401,7 @@ func (t *mintingTestHarness) assertSeedlingsMatchSprouts(
 		isCommittedBatch := func(batch *tapgarden.MintingBatch) bool {
 			return batch.State() == tapgarden.BatchStateCommitted
 		}
-		batch, err := chanutils.First(pendingBatches, isCommittedBatch)
+		batch, err := fn.First(pendingBatches, isCommittedBatch)
 		if err != nil {
 			return false
 		}
@@ -450,7 +450,7 @@ func (t *mintingTestHarness) assertGenesisPsbtFinalized() {
 	t.Helper()
 
 	// Ensure that a request to finalize the PSBt has come across.
-	_, err := chanutils.RecvOrTimeout(
+	_, err := fn.RecvOrTimeout(
 		t.wallet.SignPsbtSignal, defaultTimeout,
 	)
 	require.NoError(t, err, "psbt sign req not sent")
@@ -465,7 +465,7 @@ func (t *mintingTestHarness) assertGenesisPsbtFinalized() {
 	isNotCancelledBatch := func(batch *tapgarden.MintingBatch) bool {
 		return !isCancelledBatch(batch)
 	}
-	pendingBatch, err := chanutils.First(pendingBatches, isNotCancelledBatch)
+	pendingBatch, err := fn.First(pendingBatches, isNotCancelledBatch)
 	require.NoError(t, err)
 
 	// The minting key of the batch should match the public key
@@ -473,7 +473,7 @@ func (t *mintingTestHarness) assertGenesisPsbtFinalized() {
 	batchKey, _, err := pendingBatch.MintingOutputKey()
 	require.NoError(t, err)
 
-	importedKey, err := chanutils.RecvOrTimeout(
+	importedKey, err := fn.RecvOrTimeout(
 		t.wallet.ImportPubKeySignal, defaultTimeout,
 	)
 	require.NoError(t, err, "pubkey import req not sent")
@@ -485,7 +485,7 @@ func (t *mintingTestHarness) assertGenesisPsbtFinalized() {
 func (t *mintingTestHarness) assertTxPublished() *wire.MsgTx {
 	t.Helper()
 
-	tx, err := chanutils.RecvOrTimeout(t.chain.PublishReq, defaultTimeout)
+	tx, err := fn.RecvOrTimeout(t.chain.PublishReq, defaultTimeout)
 	require.NoError(t, err)
 
 	return *tx
@@ -497,7 +497,7 @@ func (t *mintingTestHarness) assertTxPublished() *wire.MsgTx {
 func (t *mintingTestHarness) assertConfReqSent(tx *wire.MsgTx,
 	block *wire.MsgBlock) func() {
 
-	reqNo, err := chanutils.RecvOrTimeout(
+	reqNo, err := fn.RecvOrTimeout(
 		t.chain.ConfReqSignal, defaultTimeout,
 	)
 	require.NoError(t, err)