graph+discovery: update graph/db gossip backlog interfaces to use iter.Seq2

This lets us emit a rich error if things fail when first creating the
iterator, or if any of the yield attempts fail.

Author: Roasbeef

discovery/chan_series.go

@@ -114,15 +114,15 @@ func (c *ChanSeries) UpdatesInHorizon(chain chainhash.Hash,
 	return func(yield func(lnwire.Message, error) bool) {
 		// First, we'll query for all the set of channels that have an
 		// update that falls within the specified horizon.
-		chansInHorizon, err := c.graph.ChanUpdatesInHorizon(
+		chansInHorizon := c.graph.ChanUpdatesInHorizon(
 			startTime, endTime,
 		)
-		if err != nil {
-			yield(nil, err)
-			return
-		}
 
-		for channel := range chansInHorizon {
+		for channel, err := range chansInHorizon {
+			if err != nil {
+				yield(nil, err)
+				return
+			}
 			// If the channel hasn't been fully advertised yet, or
 			// is a private channel, then we'll skip it as we can't
 			// construct a full authentication proof if one is
@@ -181,10 +181,14 @@ func (c *ChanSeries) UpdatesInHorizon(chain chainhash.Hash,
 		// Next, we'll send out all the node announcements that have an
 		// update within the horizon as well. We send these second to
 		// ensure that they follow any active channels they have.
-		nodeAnnsInHorizon, err := c.graph.NodeUpdatesInHorizon(
+		nodeAnnsInHorizon := c.graph.NodeUpdatesInHorizon(
 			startTime, endTime, graphdb.WithIterPublicNodesOnly(),
 		)
-		for nodeAnn := range nodeAnnsInHorizon {
+		for nodeAnn, err := range nodeAnnsInHorizon {
+			if err != nil {
+				yield(nil, err)
+				return
+			}
 			nodeUpdate, err := nodeAnn.NodeAnnouncement(true)
 			if err != nil {
 				if !yield(nil, err) {

graph/builder.go

@@ -593,13 +593,14 @@ func (b *Builder) pruneZombieChans() error {
 
 	startTime := time.Unix(0, 0)
 	endTime := time.Now().Add(-1 * chanExpiry)
-	oldEdgesIter, err := b.cfg.Graph.ChanUpdatesInHorizon(startTime, endTime)
-	if err != nil {
-		return fmt.Errorf("unable to fetch expired channel updates "+
-			"chans: %v", err)
-	}
+	oldEdgesIter := b.cfg.Graph.ChanUpdatesInHorizon(startTime, endTime)
+
+	for u, err := range oldEdgesIter {
+		if err != nil {
+			return fmt.Errorf("unable to fetch expired "+
+				"channel updates chans: %v", err)
+		}
 
-	for u := range oldEdgesIter {
 		err = filterPruneChans(u.Info, u.Policy1, u.Policy2)
 		if err != nil {
 			return fmt.Errorf("error filtering channels to "+
@@ -619,7 +620,7 @@ func (b *Builder) pruneZombieChans() error {
 		toPrune = append(toPrune, chanID)
 		log.Tracef("Pruning zombie channel with ChannelID(%v)", chanID)
 	}
-	err = b.cfg.Graph.DeleteChannelEdges(
+	err := b.cfg.Graph.DeleteChannelEdges(
 		b.cfg.StrictZombiePruning, true, toPrune...,
 	)
 	if err != nil {

graph/db/benchmark_test.go

@@ -15,6 +15,7 @@ import (
 	"github.com/btcsuite/btcd/chaincfg"
 	"github.com/btcsuite/btclog/v2"
 	"github.com/lightningnetwork/lnd/batch"
+	"github.com/lightningnetwork/lnd/fn/v2"
 	"github.com/lightningnetwork/lnd/graph/db/models"
 	"github.com/lightningnetwork/lnd/kvdb"
 	"github.com/lightningnetwork/lnd/kvdb/postgres"
@@ -788,9 +789,10 @@ func BenchmarkGraphReadMethods(b *testing.B) {
 		{
 			name: "NodeUpdatesInHorizon",
 			fn: func(b testing.TB, store V1Store) {
-				_, err := store.NodeUpdatesInHorizon(
+				iter := store.NodeUpdatesInHorizon(
 					time.Unix(0, 0), time.Now(),
 				)
+				_, err := fn.CollectErr(iter)
 				require.NoError(b, err)
 			},
 		},
@@ -836,9 +838,10 @@ func BenchmarkGraphReadMethods(b *testing.B) {
 		{
 			name: "ChanUpdatesInHorizon",
 			fn: func(b testing.TB, store V1Store) {
-				_, err := store.ChanUpdatesInHorizon(
+				iter := store.ChanUpdatesInHorizon(
 					time.Unix(0, 0), time.Now(),
 				)
+				_, err := fn.CollectErr(iter)
 				require.NoError(b, err)
 			},
 		},