diff --git a/graph/db/graph_test.go b/graph/db/graph_test.go index 2fd7e508b..590c077d6 100644 --- a/graph/db/graph_test.go +++ b/graph/db/graph_test.go @@ -4,7 +4,6 @@ import ( "bytes" "context" "crypto/sha256" - "database/sql" "encoding/hex" "errors" "fmt" @@ -397,19 +396,19 @@ func TestSourceNode(t *testing.T) { compareNodes(t, testNode, sourceNode) } -// TestSetSourceNodeSameTimestamp demonstrates that SetSourceNode can return an -// error when called with the same last update timestamp. Calling SetSourceNode -// with the same timestamp should be allowed (unlike AddNode), as it is -// possible that our own node announcement may change quickly. This will be -// fixed in an upcoming commit. +// TestSetSourceNodeSameTimestamp tests that SetSourceNode accepts updates +// with the same timestamp. This is necessary because multiple code paths +// (setSelfNode, createNewHiddenService, RPC updates) can race during startup, +// reading the same old timestamp and independently incrementing it to the same +// new value. For our own node, we want parameter changes to persist even with +// timestamp collisions (unlike network gossip where same timestamp means same +// content). func TestSetSourceNodeSameTimestamp(t *testing.T) { t.Parallel() ctx := t.Context() graph := MakeTestGraph(t) - _, isSQLStore := graph.V1Store.(*SQLStore) - // Create and set the initial source node. testNode := createTestVertex(t) require.NoError(t, graph.SetSourceNode(ctx, testNode)) @@ -421,8 +420,9 @@ func TestSetSourceNodeSameTimestamp(t *testing.T) { // Create a modified version of the node with the same timestamp but // different parameters (e.g., different alias and color). This - // could well be the case for our own node announcement (unlike other - // announcements where same timestamp means same parameters). + // simulates the race condition where multiple goroutines read the + // same old timestamp, independently increment it, and try to update + // with different changes. modifiedNode := models.NewV1Node( testNode.PubKeyBytes, &models.NodeV1Fields{ // Same timestamp. @@ -437,20 +437,20 @@ func TestSetSourceNodeSameTimestamp(t *testing.T) { ) // Attempt to set the source node with the same timestamp but - // different parameters. - err = graph.SetSourceNode(ctx, modifiedNode) + // different parameters. This should now succeed for both SQL and KV + // stores. The SQL store uses UpsertSourceNode which removes the + // strict timestamp constraint, allowing last-write-wins semantics. + require.NoError(t, graph.SetSourceNode(ctx, modifiedNode)) - // The SQL store will return sql.ErrNoRows because the UPDATE clause - // in the upsert query requires the new timestamp to be strictly - // greater than the existing one. When this condition is not met, no - // rows are updated and the SQL query returns ErrNoRows. The bbolt KV - // store, on the other hand, silently ignores stale updates and returns - // no error. - if isSQLStore { - require.ErrorIs(t, err, sql.ErrNoRows) - } else { - require.NoError(t, err) - } + // Verify that the parameter changes actually persisted. + updatedNode, err := graph.SourceNode(ctx) + require.NoError(t, err) + require.Equal(t, "different-alias", updatedNode.Alias.UnwrapOr("")) + require.Equal( + t, color.RGBA{R: 100, G: 200, B: 50, A: 0}, + updatedNode.Color.UnwrapOr(color.RGBA{}), + ) + require.Equal(t, testNode.LastUpdate, updatedNode.LastUpdate) } // TestEdgeInsertionDeletion tests the basic CRUD operations for channel edges. diff --git a/graph/db/sql_store.go b/graph/db/sql_store.go index 3578fd945..650770726 100644 --- a/graph/db/sql_store.go +++ b/graph/db/sql_store.go @@ -42,6 +42,7 @@ type SQLQueries interface { Node queries. */ UpsertNode(ctx context.Context, arg sqlc.UpsertNodeParams) (int64, error) + UpsertSourceNode(ctx context.Context, arg sqlc.UpsertSourceNodeParams) (int64, error) GetNodeByPubKey(ctx context.Context, arg sqlc.GetNodeByPubKeyParams) (sqlc.GraphNode, error) GetNodesByIDs(ctx context.Context, ids []int64) ([]sqlc.GraphNode, error) GetNodeIDByPubKey(ctx context.Context, arg sqlc.GetNodeIDByPubKeyParams) (int64, error) @@ -521,7 +522,14 @@ func (s *SQLStore) SetSourceNode(ctx context.Context, node *models.Node) error { return s.db.ExecTx(ctx, sqldb.WriteTxOpt(), func(db SQLQueries) error { - id, err := upsertNode(ctx, db, node) + // For the source node, we use a less strict upsert that allows + // updates even when the timestamp hasn't changed. This handles + // the race condition where multiple goroutines (e.g., + // setSelfNode, createNewHiddenService, RPC updates) read the + // same old timestamp, independently increment it, and try to + // write concurrently. We want all parameter changes to persist, + // even if timestamps collide. + id, err := upsertSourceNode(ctx, db, node) if err != nil { return fmt.Errorf("unable to upsert source node: %w", err) @@ -3599,6 +3607,154 @@ func getNodeFeatures(ctx context.Context, db SQLQueries, return features, nil } +// upsertNodeAncillaryData updates the node's features, addresses, and extra +// signed fields. This is common logic shared by upsertNode and +// upsertSourceNode. +func upsertNodeAncillaryData(ctx context.Context, db SQLQueries, + nodeID int64, node *models.Node) error { + + // Update the node's features. + err := upsertNodeFeatures(ctx, db, nodeID, node.Features) + if err != nil { + return fmt.Errorf("inserting node features: %w", err) + } + + // Update the node's addresses. + err = upsertNodeAddresses(ctx, db, nodeID, node.Addresses) + if err != nil { + return fmt.Errorf("inserting node addresses: %w", err) + } + + // Convert the flat extra opaque data into a map of TLV types to + // values. + extra, err := marshalExtraOpaqueData(node.ExtraOpaqueData) + if err != nil { + return fmt.Errorf("unable to marshal extra opaque data: %w", + err) + } + + // Update the node's extra signed fields. + err = upsertNodeExtraSignedFields(ctx, db, nodeID, extra) + if err != nil { + return fmt.Errorf("inserting node extra TLVs: %w", err) + } + + return nil +} + +// populateNodeParams populates the common node parameters from a models.Node. +// This is a helper for building UpsertNodeParams and UpsertSourceNodeParams. +func populateNodeParams(node *models.Node, + setParams func(lastUpdate sql.NullInt64, alias, + colorStr sql.NullString, signature []byte)) error { + + if !node.HaveAnnouncement() { + return nil + } + + switch node.Version { + case lnwire.GossipVersion1: + lastUpdate := sqldb.SQLInt64(node.LastUpdate.Unix()) + var alias, colorStr sql.NullString + + node.Color.WhenSome(func(rgba color.RGBA) { + colorStr = sqldb.SQLStrValid(EncodeHexColor(rgba)) + }) + node.Alias.WhenSome(func(s string) { + alias = sqldb.SQLStrValid(s) + }) + + setParams(lastUpdate, alias, colorStr, node.AuthSigBytes) + + case lnwire.GossipVersion2: + // No-op for now. + + default: + return fmt.Errorf("unknown gossip version: %d", node.Version) + } + + return nil +} + +// buildNodeUpsertParams builds the parameters for upserting a node using the +// strict UpsertNode query (requires timestamp to be increasing). +func buildNodeUpsertParams(node *models.Node) (sqlc.UpsertNodeParams, error) { + params := sqlc.UpsertNodeParams{ + Version: int16(lnwire.GossipVersion1), + PubKey: node.PubKeyBytes[:], + } + + err := populateNodeParams( + node, func(lastUpdate sql.NullInt64, alias, + colorStr sql.NullString, + signature []byte) { + + params.LastUpdate = lastUpdate + params.Alias = alias + params.Color = colorStr + params.Signature = signature + }) + + return params, err +} + +// buildSourceNodeUpsertParams builds the parameters for upserting the source +// node using the lenient UpsertSourceNode query (allows same timestamp). +func buildSourceNodeUpsertParams(node *models.Node) ( + sqlc.UpsertSourceNodeParams, error) { + + params := sqlc.UpsertSourceNodeParams{ + Version: int16(lnwire.GossipVersion1), + PubKey: node.PubKeyBytes[:], + } + + err := populateNodeParams( + node, func(lastUpdate sql.NullInt64, alias, + colorStr sql.NullString, signature []byte) { + + params.LastUpdate = lastUpdate + params.Alias = alias + params.Color = colorStr + params.Signature = signature + }, + ) + + return params, err +} + +// upsertSourceNode upserts the source node record into the database using a +// less strict upsert that allows updates even when the timestamp hasn't +// changed. This is necessary to handle concurrent updates to our own node +// during startup and runtime. The node's features, addresses and extra TLV +// types are also updated. The node's DB ID is returned. +func upsertSourceNode(ctx context.Context, db SQLQueries, + node *models.Node) (int64, error) { + + params, err := buildSourceNodeUpsertParams(node) + if err != nil { + return 0, err + } + + nodeID, err := db.UpsertSourceNode(ctx, params) + if err != nil { + return 0, fmt.Errorf("upserting source node(%x): %w", + node.PubKeyBytes, err) + } + + // We can exit here if we don't have the announcement yet. + if !node.HaveAnnouncement() { + return nodeID, nil + } + + // Update the ancillary node data (features, addresses, extra fields). + err = upsertNodeAncillaryData(ctx, db, nodeID, node) + if err != nil { + return 0, err + } + + return nodeID, nil +} + // upsertNode upserts the node record into the database. If the node already // exists, then the node's information is updated. If the node doesn't exist, // then a new node is created. The node's features, addresses and extra TLV @@ -3606,33 +3762,9 @@ func getNodeFeatures(ctx context.Context, db SQLQueries, func upsertNode(ctx context.Context, db SQLQueries, node *models.Node) (int64, error) { - params := sqlc.UpsertNodeParams{ - Version: int16(lnwire.GossipVersion1), - PubKey: node.PubKeyBytes[:], - } - - if node.HaveAnnouncement() { - switch node.Version { - case lnwire.GossipVersion1: - params.LastUpdate = sqldb.SQLInt64( - node.LastUpdate.Unix(), - ) - - case lnwire.GossipVersion2: - - default: - return 0, fmt.Errorf("unknown gossip version: %d", - node.Version) - } - - node.Color.WhenSome(func(rgba color.RGBA) { - params.Color = sqldb.SQLStrValid(EncodeHexColor(rgba)) - }) - node.Alias.WhenSome(func(s string) { - params.Alias = sqldb.SQLStrValid(s) - }) - - params.Signature = node.AuthSigBytes + params, err := buildNodeUpsertParams(node) + if err != nil { + return 0, err } nodeID, err := db.UpsertNode(ctx, params) @@ -3646,30 +3778,10 @@ func upsertNode(ctx context.Context, db SQLQueries, return nodeID, nil } - // Update the node's features. - err = upsertNodeFeatures(ctx, db, nodeID, node.Features) + // Update the ancillary node data (features, addresses, extra fields). + err = upsertNodeAncillaryData(ctx, db, nodeID, node) if err != nil { - return 0, fmt.Errorf("inserting node features: %w", err) - } - - // Update the node's addresses. - err = upsertNodeAddresses(ctx, db, nodeID, node.Addresses) - if err != nil { - return 0, fmt.Errorf("inserting node addresses: %w", err) - } - - // Convert the flat extra opaque data into a map of TLV types to - // values. - extra, err := marshalExtraOpaqueData(node.ExtraOpaqueData) - if err != nil { - return 0, fmt.Errorf("unable to marshal extra opaque data: %w", - err) - } - - // Update the node's extra signed fields. - err = upsertNodeExtraSignedFields(ctx, db, nodeID, extra) - if err != nil { - return 0, fmt.Errorf("inserting node extra TLVs: %w", err) + return 0, err } return nodeID, nil