Merge pull request #9869 from ellemouton/graphSQL8-channels-schema

sqldb+graph/db: add channel tables and implement some channel CRUD

Author: guggero

graph/db/graph_test.go

@@ -82,7 +82,6 @@ func createLightningNode(priv *btcec.PrivateKey) *models.LightningNode {
 		Alias:                "kek" + hex.EncodeToString(pub),
 		Features:             testFeatures,
 		Addresses:            testAddrs,
-		ExtraOpaqueData:      make([]byte, 0),
 	}
 	copy(n.PubKeyBytes[:], priv.PubKey().SerializeCompressed())
 
@@ -262,7 +261,7 @@ func TestNodeInsertionAndDeletion(t *testing.T) {
 func TestPartialNode(t *testing.T) {
 	t.Parallel()
 
-	graph := MakeTestGraph(t)
+	graph := MakeTestGraphNew(t)
 
 	// To insert a partial node, we need to add a channel edge that has
 	// node keys for nodes we are not yet aware
@@ -1862,7 +1861,7 @@ func TestGraphPruning(t *testing.T) {
 func TestHighestChanID(t *testing.T) {
 	t.Parallel()
 
-	graph := MakeTestGraph(t)
+	graph := MakeTestGraphNew(t)
 
 	// If we don't yet have any channels in the database, then we should
 	// get a channel ID of zero if we ask for the highest channel ID.
@@ -3333,34 +3332,33 @@ func TestPruneGraphNodes(t *testing.T) {
 func TestAddChannelEdgeShellNodes(t *testing.T) {
 	t.Parallel()
 
-	graph := MakeTestGraph(t)
+	graph := MakeTestGraphNew(t)
 
 	// To start, we'll create two nodes, and only add one of them to the
 	// channel graph.
 	node1 := createTestVertex(t)
-	if err := graph.AddLightningNode(node1); err != nil {
-		t.Fatalf("unable to add node: %v", err)
-	}
+	require.NoError(t, graph.SetSourceNode(node1))
 	node2 := createTestVertex(t)
 
 	// We'll now create an edge between the two nodes, as a result, node2
 	// should be inserted into the database as a shell node.
 	edgeInfo, _ := createEdge(100, 0, 0, 0, node1, node2)
-	if err := graph.AddChannelEdge(&edgeInfo); err != nil {
-		t.Fatalf("unable to add edge: %v", err)
-	}
+	require.NoError(t, graph.AddChannelEdge(&edgeInfo))
 
 	// Ensure that node1 was inserted as a full node, while node2 only has
 	// a shell node present.
 	node1, err := graph.FetchLightningNode(node1.PubKeyBytes)
 	require.NoError(t, err, "unable to fetch node1")
-	if !node1.HaveNodeAnnouncement {
-		t.Fatalf("have shell announcement for node1, shouldn't")
-	}
+	require.True(t, node1.HaveNodeAnnouncement)
 
 	node2, err = graph.FetchLightningNode(node2.PubKeyBytes)
 	require.NoError(t, err, "unable to fetch node2")
 	require.False(t, node2.HaveNodeAnnouncement)
+
+	// Show that attempting to add the channel again will result in an
+	// error.
+	err = graph.AddChannelEdge(&edgeInfo)
+	require.ErrorIs(t, err, ErrEdgeAlreadyExist)
 }
 
 // TestNodePruningUpdateIndexDeletion tests that once a node has been removed

graph/db/kv_store.go

@@ -4122,7 +4122,7 @@ func deserializeLightningNode(r io.Reader) (models.LightningNode, error) {
 
 	// We'll try and see if there are any opaque bytes left, if not, then
 	// we'll ignore the EOF error and return the node as is.
-	node.ExtraOpaqueData, err = wire.ReadVarBytes(
+	extraBytes, err := wire.ReadVarBytes(
 		r, 0, MaxAllowedExtraOpaqueBytes, "blob",
 	)
 	switch {
@@ -4132,6 +4132,10 @@ func deserializeLightningNode(r io.Reader) (models.LightningNode, error) {
 		return models.LightningNode{}, err
 	}
 
+	if len(extraBytes) > 0 {
+		node.ExtraOpaqueData = extraBytes
+	}
+
 	return node, nil
 }
 

graph/db/sql_store.go

@@ -69,6 +69,16 @@ type SQLQueries interface {
 	*/
 	AddSourceNode(ctx context.Context, nodeID int64) error
 	GetSourceNodesByVersion(ctx context.Context, version int16) ([]sqlc.GetSourceNodesByVersionRow, error)
+
+	/*
+		Channel queries.
+	*/
+	CreateChannel(ctx context.Context, arg sqlc.CreateChannelParams) (int64, error)
+	GetChannelBySCID(ctx context.Context, arg sqlc.GetChannelBySCIDParams) (sqlc.Channel, error)
+	HighestSCID(ctx context.Context, version int16) ([]byte, error)
+
+	CreateChannelExtraType(ctx context.Context, arg sqlc.CreateChannelExtraTypeParams) error
+	InsertChannelFeature(ctx context.Context, arg sqlc.InsertChannelFeatureParams) error
 }
 
 // BatchedSQLQueries is a version of SQLQueries that's capable of batched
@@ -455,6 +465,83 @@ func (s *SQLStore) NodeUpdatesInHorizon(startTime,
 	return nodes, nil
 }
 
+// AddChannelEdge adds a new (undirected, blank) edge to the graph database. An
+// undirected edge from the two target nodes are created. The information stored
+// denotes the static attributes of the channel, such as the channelID, the keys
+// involved in creation of the channel, and the set of features that the channel
+// supports. The chanPoint and chanID are used to uniquely identify the edge
+// globally within the database.
+//
+// NOTE: part of the V1Store interface.
+func (s *SQLStore) AddChannelEdge(edge *models.ChannelEdgeInfo,
+	opts ...batch.SchedulerOption) error {
+
+	ctx := context.TODO()
+
+	var alreadyExists bool
+	r := &batch.Request[SQLQueries]{
+		Opts: batch.NewSchedulerOptions(opts...),
+		Reset: func() {
+			alreadyExists = false
+		},
+		Do: func(tx SQLQueries) error {
+			err := insertChannel(ctx, tx, edge)
+
+			// Silence ErrEdgeAlreadyExist so that the batch can
+			// succeed, but propagate the error via local state.
+			if errors.Is(err, ErrEdgeAlreadyExist) {
+				alreadyExists = true
+				return nil
+			}
+
+			return err
+		},
+		OnCommit: func(err error) error {
+			switch {
+			case err != nil:
+				return err
+			case alreadyExists:
+				return ErrEdgeAlreadyExist
+			default:
+				s.rejectCache.remove(edge.ChannelID)
+				s.chanCache.remove(edge.ChannelID)
+				return nil
+			}
+		},
+	}
+
+	return s.chanScheduler.Execute(ctx, r)
+}
+
+// HighestChanID returns the "highest" known channel ID in the channel graph.
+// This represents the "newest" channel from the PoV of the chain. This method
+// can be used by peers to quickly determine if their graphs are in sync.
+//
+// NOTE: This is part of the V1Store interface.
+func (s *SQLStore) HighestChanID() (uint64, error) {
+	ctx := context.TODO()
+
+	var highestChanID uint64
+	err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error {
+		chanID, err := db.HighestSCID(ctx, int16(ProtocolV1))
+		if errors.Is(err, sql.ErrNoRows) {
+			return nil
+		} else if err != nil {
+			return fmt.Errorf("unable to fetch highest chan ID: %w",
+				err)
+		}
+
+		highestChanID = byteOrder.Uint64(chanID)
+
+		return nil
+	}, sqldb.NoOpReset)
+	if err != nil {
+		return 0, fmt.Errorf("unable to fetch highest chan ID: %w", err)
+	}
+
+	return highestChanID, nil
+}
+
 // getNodeByPubKey attempts to look up a target node by its public key.
 func getNodeByPubKey(ctx context.Context, db SQLQueries,
 	pubKey route.Vertex) (int64, *models.LightningNode, error) {
@@ -494,10 +581,9 @@ func buildNode(ctx context.Context, db SQLQueries, dbNode *sqlc.Node) (
 	copy(pub[:], dbNode.PubKey)
 
 	node := &models.LightningNode{
-		PubKeyBytes:     pub,
-		Features:        lnwire.EmptyFeatureVector(),
-		LastUpdate:      time.Unix(0, 0),
-		ExtraOpaqueData: make([]byte, 0),
+		PubKeyBytes: pub,
+		Features:    lnwire.EmptyFeatureVector(),
+		LastUpdate:  time.Unix(0, 0),
 	}
 
 	if len(dbNode.Signature) == 0 {
@@ -1023,3 +1109,151 @@ func marshalExtraOpaqueData(data []byte) (map[uint64][]byte, error) {
 
 	return records, nil
 }
+
+// insertChannel inserts a new channel record into the database.
+func insertChannel(ctx context.Context, db SQLQueries,
+	edge *models.ChannelEdgeInfo) error {
+
+	var chanIDB [8]byte
+	byteOrder.PutUint64(chanIDB[:], edge.ChannelID)
+
+	// Make sure that the channel doesn't already exist. We do this
+	// explicitly instead of relying on catching a unique constraint error
+	// because relying on SQL to throw that error would abort the entire
+	// batch of transactions.
+	_, err := db.GetChannelBySCID(
+		ctx, sqlc.GetChannelBySCIDParams{
+			Scid:    chanIDB[:],
+			Version: int16(ProtocolV1),
+		},
+	)
+	if err == nil {
+		return ErrEdgeAlreadyExist
+	} else if !errors.Is(err, sql.ErrNoRows) {
+		return fmt.Errorf("unable to fetch channel: %w", err)
+	}
+
+	// Make sure that at least a "shell" entry for each node is present in
+	// the nodes table.
+	node1DBID, err := maybeCreateShellNode(ctx, db, edge.NodeKey1Bytes)
+	if err != nil {
+		return fmt.Errorf("unable to create shell node: %w", err)
+	}
+
+	node2DBID, err := maybeCreateShellNode(ctx, db, edge.NodeKey2Bytes)
+	if err != nil {
+		return fmt.Errorf("unable to create shell node: %w", err)
+	}
+
+	var capacity sql.NullInt64
+	if edge.Capacity != 0 {
+		capacity = sqldb.SQLInt64(int64(edge.Capacity))
+	}
+
+	createParams := sqlc.CreateChannelParams{
+		Version:     int16(ProtocolV1),
+		Scid:        chanIDB[:],
+		NodeID1:     node1DBID,
+		NodeID2:     node2DBID,
+		Outpoint:    edge.ChannelPoint.String(),
+		Capacity:    capacity,
+		BitcoinKey1: edge.BitcoinKey1Bytes[:],
+		BitcoinKey2: edge.BitcoinKey2Bytes[:],
+	}
+
+	if edge.AuthProof != nil {
+		proof := edge.AuthProof
+
+		createParams.Node1Signature = proof.NodeSig1Bytes
+		createParams.Node2Signature = proof.NodeSig2Bytes
+		createParams.Bitcoin1Signature = proof.BitcoinSig1Bytes
+		createParams.Bitcoin2Signature = proof.BitcoinSig2Bytes
+	}
+
+	// Insert the new channel record.
+	dbChanID, err := db.CreateChannel(ctx, createParams)
+	if err != nil {
+		return err
+	}
+
+	// Insert any channel features.
+	if len(edge.Features) != 0 {
+		chanFeatures := lnwire.NewRawFeatureVector()
+		err := chanFeatures.Decode(bytes.NewReader(edge.Features))
+		if err != nil {
+			return err
+		}
+
+		fv := lnwire.NewFeatureVector(chanFeatures, lnwire.Features)
+		for feature := range fv.Features() {
+			err = db.InsertChannelFeature(
+				ctx, sqlc.InsertChannelFeatureParams{
+					ChannelID:  dbChanID,
+					FeatureBit: int32(feature),
+				},
+			)
+			if err != nil {
+				return fmt.Errorf("unable to insert "+
+					"channel(%d) feature(%v): %w", dbChanID,
+					feature, err)
+			}
+		}
+	}
+
+	// Finally, insert any extra TLV fields in the channel announcement.
+	extra, err := marshalExtraOpaqueData(edge.ExtraOpaqueData)
+	if err != nil {
+		return fmt.Errorf("unable to marshal extra opaque data: %w",
+			err)
+	}
+
+	for tlvType, value := range extra {
+		err := db.CreateChannelExtraType(
+			ctx, sqlc.CreateChannelExtraTypeParams{
+				ChannelID: dbChanID,
+				Type:      int64(tlvType),
+				Value:     value,
+			},
+		)
+		if err != nil {
+			return fmt.Errorf("unable to upsert channel(%d) extra "+
+				"signed field(%v): %w", edge.ChannelID,
+				tlvType, err)
+		}
+	}
+
+	return nil
+}
+
+// maybeCreateShellNode checks if a shell node entry exists for the
+// given public key. If it does not exist, then a new shell node entry is
+// created. The ID of the node is returned. A shell node only has a protocol
+// version and public key persisted.
+func maybeCreateShellNode(ctx context.Context, db SQLQueries,
+	pubKey route.Vertex) (int64, error) {
+
+	dbNode, err := db.GetNodeByPubKey(
+		ctx, sqlc.GetNodeByPubKeyParams{
+			PubKey:  pubKey[:],
+			Version: int16(ProtocolV1),
+		},
+	)
+	// The node exists. Return the ID.
+	if err == nil {
+		return dbNode.ID, nil
+	} else if !errors.Is(err, sql.ErrNoRows) {
+		return 0, err
+	}
+
+	// Otherwise, the node does not exist, so we create a shell entry for
+	// it.
+	id, err := db.UpsertNode(ctx, sqlc.UpsertNodeParams{
+		Version: int16(ProtocolV1),
+		PubKey:  pubKey[:],
+	})
+	if err != nil {
+		return 0, fmt.Errorf("unable to create shell node: %w", err)
+	}
+
+	return id, nil
+}