Merge pull request #626 from lightninglabs/universe-cache

multi: add new caches for universe proofs, roots, keys, and config

Author: Roasbeef

config.go

@@ -17,6 +17,7 @@ import (
 	"github.com/lightningnetwork/lnd"
 	"github.com/lightningnetwork/lnd/build"
 	"github.com/lightningnetwork/lnd/signal"
+	"golang.org/x/time/rate"
 	"google.golang.org/grpc"
 )
 
@@ -129,6 +130,15 @@ type Config struct {
 	// This applies to federation syncing as well as RPC insert and query.
 	UniversePublicAccess bool
 
+	// UniverseQueriesPerSecond is the maximum number of queries per
+	// second across the set of active universe queries that is permitted.
+	// Anything above this starts to get rate limited.
+	UniverseQueriesPerSecond rate.Limit
+
+	// UniverseQueriesBurst is the burst budget for the universe query rate
+	// limiting.
+	UniverseQueriesBurst int
+
 	Prometheus monitoring.PrometheusConfig
 
 	// LogWriter is the root logger that all of the daemon's subloggers are

go.mod

@@ -39,6 +39,7 @@ require (
 	golang.org/x/net v0.17.0
 	golang.org/x/sync v0.2.0
 	golang.org/x/term v0.13.0
+	golang.org/x/time v0.0.0-20220224211638-0e9765cccd65
 	google.golang.org/grpc v1.56.3
 	google.golang.org/protobuf v1.30.0
 	gopkg.in/macaroon-bakery.v2 v2.1.0
@@ -187,7 +188,6 @@ require (
 	golang.org/x/mod v0.10.0 // indirect
 	golang.org/x/sys v0.13.0 // indirect
 	golang.org/x/text v0.13.0 // indirect
-	golang.org/x/time v0.0.0-20220224211638-0e9765cccd65 // indirect
 	golang.org/x/tools v0.9.1 // indirect
 	google.golang.org/genproto v0.0.0-20230410155749-daa745c078e1 // indirect
 	gopkg.in/errgo.v1 v1.0.1 // indirect

itest/addrs_test.go

@@ -239,7 +239,7 @@ func testAddressAssetSyncer(t *harnessTest) {
 	require.NoError(t.t, err)
 
 	// Bob's Universe stats should show that he now has two assets.
-	AssertUniverseStats(t.t, bob, 2, 0, 2, 1)
+	AssertUniverseStats(t.t, bob, 2, 2, 1)
 
 	// Bob should not be able to make an address for a random asset ID
 	// that he nor Alice are aware of.
@@ -344,7 +344,7 @@ func testAddressAssetSyncer(t *harnessTest) {
 			},
 		)
 		require.NoError(t.t, err)
-		AssertUniverseStats(t.t, bob, 0, 0, 0, 0)
+		AssertUniverseStats(t.t, bob, 0, 0, 0)
 	}
 
 	// If we restart Bob with the syncer disabled and no automatic sync
@@ -401,11 +401,11 @@ func testAddressAssetSyncer(t *harnessTest) {
 
 	// Bob's Universe stats should show that he has now synced both assets
 	// from the second mint and the single asset group from that mint.
-	AssertUniverseStats(t.t, bob, 2, 0, 2, 1)
+	AssertUniverseStats(t.t, bob, 2, 2, 1)
 
 	// Alice's Universe stats should reflect the extra syncs from the asset
 	// group lookups by Bob.
-	AssertUniverseStats(t.t, t.tapd, 4, 4, 4, 2)
+	AssertUniverseStats(t.t, t.tapd, 4, 4, 2)
 
 	// If Alice now mints a reissuance for the second asset group, Bob
 	// should successfully sync that new asset.

itest/assertions.go

@@ -1226,7 +1226,7 @@ func AssertUniverseKeysEqual(t *testing.T, uniIDs []*unirpc.ID,
 }
 
 func AssertUniverseStats(t *testing.T, client unirpc.UniverseClient,
-	numProofs, numSyncs, numAssets, numGroups int) {
+	numProofs, numAssets, numGroups int) {
 
 	err := wait.NoError(func() error {
 		uniStats, err := client.UniverseStats(
@@ -1240,10 +1240,6 @@ func AssertUniverseStats(t *testing.T, client unirpc.UniverseClient,
 			return fmt.Errorf("expected %v proofs, got %v",
 				numProofs, uniStats.NumTotalProofs)
 		}
-		if numSyncs != int(uniStats.NumTotalSyncs) {
-			return fmt.Errorf("expected %v syncs, got %v",
-				numSyncs, uniStats.NumTotalSyncs)
-		}
 		if numAssets != int(uniStats.NumTotalAssets) {
 			return fmt.Errorf("expected %v assets, got %v",
 				numAssets, uniStats.NumTotalAssets)

itest/universe_test.go

@@ -453,7 +453,7 @@ func testUniverseFederation(t *harnessTest) {
 
 	// Bob's Universe stats should show that he now has a single asset. We
 	// should also be able to query for stats specifically for the asset.
-	AssertUniverseStats(t.t, bob, 1, 0, 1, 0)
+	AssertUniverseStats(t.t, bob, 1, 1, 0)
 
 	// Test the content of the universe info call.
 	info, err := bob.Info(ctxt, &unirpc.InfoRequest{})
@@ -545,7 +545,7 @@ func testUniverseFederation(t *harnessTest) {
 
 	// Bob's stats should also now show that there're three total asset as
 	// well as three proofs.
-	AssertUniverseStats(t.t, bob, 3, 0, 3, 1)
+	AssertUniverseStats(t.t, bob, 3, 3, 1)
 
 	// We should be able to find both the new assets in the set of universe
 	// stats for an asset.

proof/proof.go

@@ -446,3 +446,29 @@ func (p *Proof) IsUnknownVersion() bool {
 		return true
 	}
 }
+
+// SparseDecode can be used to decode a proof from a reader without decoding
+// and parsing the entire thing. This handles ignoring the magic bytes, and
+// will decode directly into the target records.
+func SparseDecode(r io.Reader, records ...tlv.Record) error {
+	// The very first byte of the serialized proof is a magic byte, so
+	// we'll read one byte to skip it.
+	var magicBytes [PrefixMagicBytesLength]byte
+	_, err := r.Read(magicBytes[:])
+	if err != nil {
+		return err
+	}
+
+	if magicBytes != PrefixMagicBytes {
+		return fmt.Errorf("invalid prefix magic bytes, expected %s, "+
+			"got %s", string(PrefixMagicBytes[:]),
+			string(magicBytes[:]))
+	}
+
+	proofStream, err := tlv.NewStream(records...)
+	if err != nil {
+		return err
+	}
+
+	return proofStream.Decode(r)
+}

rpcserver.go

@@ -46,6 +46,7 @@ import (
 	"github.com/lightningnetwork/lnd/lnrpc"
 	"github.com/lightningnetwork/lnd/lnwallet/chainfee"
 	"github.com/lightningnetwork/lnd/signal"
+	"golang.org/x/time/rate"
 	"google.golang.org/grpc"
 )
 
@@ -116,6 +117,8 @@ type rpcServer struct {
 
 	blockTimestampCache *lru.Cache[uint32, cacheableTimestamp]
 
+	proofQueryRateLimiter *rate.Limiter
+
 	quit chan struct{}
 	wg   sync.WaitGroup
 }
@@ -132,7 +135,10 @@ func newRPCServer(interceptor signal.Interceptor,
 			maxNumBlocksInCache,
 		),
 		quit: make(chan struct{}),
-		cfg:  cfg,
+		proofQueryRateLimiter: rate.NewLimiter(
+			cfg.UniverseQueriesPerSecond, cfg.UniverseQueriesBurst,
+		),
+		cfg: cfg,
 	}, nil
 }
 
@@ -2897,6 +2903,12 @@ func marshalUniverseRoot(node universe.Root) (*unirpc.UniverseRoot, error) {
 func (r *rpcServer) AssetRoots(ctx context.Context,
 	req *unirpc.AssetRootRequest) (*unirpc.AssetRootResponse, error) {
 
+	// Check the rate limiter to see if we need to wait at all. If not then
+	// this'll be a noop.
+	if err := r.proofQueryRateLimiter.Wait(ctx); err != nil {
+		return nil, err
+	}
+
 	// First, we'll retrieve the full set of known asset Universe roots.
 	assetRoots, err := r.cfg.UniverseArchive.RootNodes(
 		ctx, universe.RootNodesQuery{
@@ -3083,6 +3095,12 @@ func (r *rpcServer) QueryAssetRoots(ctx context.Context,
 			"given universe")
 	}
 
+	// Check the rate limiter to see if we need to wait at all. If not then
+	// this'll be a noop.
+	if err := r.proofQueryRateLimiter.Wait(ctx); err != nil {
+		return nil, err
+	}
+
 	// Query for both a issaunce and transfer universe root.
 	assetRoots, err := r.queryAssetProofRoots(ctx, universeID)
 	if err != nil {
@@ -3266,13 +3284,22 @@ func (r *rpcServer) AssetLeafKeys(ctx context.Context,
 		return nil, err
 	}
 
-	// TODO(roasbeef): tell above if was tring or not, then would set
-	// below diff
+	// If the proof type wasn't speciifed, then we'll return an error as we
+	// don't know which keys to actually fetch.
+	if universeID.ProofType == universe.ProofTypeUnspecified {
+		return nil, fmt.Errorf("proof type must be specified")
+	}
 
 	if req.Limit > universe.MaxPageSize || req.Limit < 0 {
 		return nil, fmt.Errorf("invalid request limit")
 	}
 
+	// Check the rate limiter to see if we need to wait at all. If not then
+	// this'll be a noop.
+	if err := r.proofQueryRateLimiter.Wait(ctx); err != nil {
+		return nil, err
+	}
+
 	leafKeys, err := r.cfg.UniverseArchive.UniverseLeafKeys(
 		ctx, universe.UniverseLeafKeysQuery{
 			Id:            universeID,
@@ -3299,23 +3326,16 @@ func (r *rpcServer) AssetLeafKeys(ctx context.Context,
 func marshalAssetLeaf(ctx context.Context, keys taprpc.KeyLookup,
 	assetLeaf *universe.Leaf) (*unirpc.AssetLeaf, error) {
 
-	// In order to display the full asset, we'll also encode the genesis
-	// proof.
-	var buf bytes.Buffer
-	if err := assetLeaf.Proof.Encode(&buf); err != nil {
-		return nil, err
-	}
-
 	rpcAsset, err := taprpc.MarshalAsset(
-		ctx, &assetLeaf.Proof.Asset, false, true, keys,
+		ctx, assetLeaf.Asset, false, true, keys,
 	)
 	if err != nil {
 		return nil, err
 	}
 
 	return &unirpc.AssetLeaf{
 		Asset: rpcAsset,
-		Proof: buf.Bytes(),
+		Proof: assetLeaf.RawProof,
 	}, nil
 }
 
@@ -3339,6 +3359,12 @@ func (r *rpcServer) AssetLeaves(ctx context.Context,
 		return nil, err
 	}
 
+	// Check the rate limiter to see if we need to wait at all. If not then
+	// this'll be a noop.
+	if err := r.proofQueryRateLimiter.Wait(ctx); err != nil {
+		return nil, err
+	}
+
 	assetLeaves, err := r.cfg.UniverseArchive.MintingLeaves(ctx, universeID)
 	if err != nil {
 		return nil, err
@@ -3533,7 +3559,7 @@ func (r *rpcServer) QueryProof(ctx context.Context,
 		return nil, err
 	}
 
-	rpcsLog.Debugf("[QueryProof]: fetching proof at (universeID=%v, "+
+	rpcsLog.Tracef("[QueryProof]: fetching proof at (universeID=%v, "+
 		"leafKey=%x)", universeID, leafKey.UniverseKey())
 
 	// Retrieve proof export config for the given universe.
@@ -3576,6 +3602,12 @@ func (r *rpcServer) QueryProof(ctx context.Context,
 			"given universe")
 	}
 
+	// Check the rate limiter to see if we need to wait at all. If not then
+	// this'll be a noop.
+	if err := r.proofQueryRateLimiter.Wait(ctx); err != nil {
+		return nil, err
+	}
+
 	// Attempt to retrieve the proof given the candidate set of universe
 	// IDs.
 	var proofs []*universe.Proof
@@ -3610,7 +3642,7 @@ func (r *rpcServer) QueryProof(ctx context.Context,
 	// not be fully specified
 	proof := proofs[0]
 
-	rpcsLog.Debugf("[QueryProof]: found proof at (universeID=%v, "+
+	rpcsLog.Tracef("[QueryProof]: found proof at (universeID=%v, "+
 		"leafKey=%x)", universeID, leafKey.UniverseKey())
 
 	return r.marshalUniverseProofLeaf(ctx, req, proof)
@@ -3635,8 +3667,9 @@ func unmarshalAssetLeaf(leaf *unirpc.AssetLeaf) (*universe.Leaf, error) {
 			Genesis:  assetProof.Asset.Genesis,
 			GroupKey: assetProof.Asset.GroupKey,
 		},
-		Proof: &assetProof,
-		Amt:   assetProof.Asset.Amount,
+		RawProof: leaf.Proof,
+		Asset:    &assetProof.Asset,
+		Amt:      assetProof.Asset.Amount,
 	}, nil
 }
 
@@ -3668,8 +3701,8 @@ func (r *rpcServer) InsertProof(ctx context.Context,
 	// If universe proof type unspecified, set based on the provided asset
 	// proof.
 	if universeID.ProofType == universe.ProofTypeUnspecified {
-		universeID.ProofType, err = universe.NewProofTypeFromAssetProof(
-			assetLeaf.Proof,
+		universeID.ProofType, err = universe.NewProofTypeFromAsset(
+			assetLeaf.Asset,
 		)
 		if err != nil {
 			return nil, err
@@ -3678,7 +3711,7 @@ func (r *rpcServer) InsertProof(ctx context.Context,
 
 	// Ensure that the new proof is of the correct type for the target
 	// universe.
-	err = universe.ValidateProofUniverseType(assetLeaf.Proof, universeID)
+	err = universe.ValidateProofUniverseType(assetLeaf.Asset, universeID)
 	if err != nil {
 		return nil, err
 	}
@@ -3694,6 +3727,12 @@ func (r *rpcServer) InsertProof(ctx context.Context,
 			"given universe")
 	}
 
+	// Check the rate limiter to see if we need to wait at all. If not then
+	// this'll be a noop.
+	if err := r.proofQueryRateLimiter.Wait(ctx); err != nil {
+		return nil, err
+	}
+
 	rpcsLog.Debugf("[InsertProof]: inserting proof at "+
 		"(universeID=%v, leafKey=%x)", universeID,
 		leafKey.UniverseKey())

tapcfg/config.go

@@ -31,6 +31,7 @@ import (
 	"github.com/lightningnetwork/lnd/signal"
 	"github.com/lightningnetwork/lnd/tor"
 	"golang.org/x/net/http2"
+	"golang.org/x/time/rate"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials"
 )
@@ -113,6 +114,16 @@ const (
 	// defaultReOrgSafeDepth is the default number of confirmations we'll
 	// wait for before considering a transaction safely buried in the chain.
 	defaultReOrgSafeDepth = 6
+
+	// defaultUniverseMaxQps is the default maximum number of queries per
+	// second for the universe server. This permis 100 queries per second
+	// by default.
+	defaultUniverseMaxQps = 100
+
+	// defaultUniverseQueriesBurst is the default burst budget for the
+	// universe queries. By default we'll allow 100 qps, with a max burst
+	// of 10 queries.
+	defaultUniverseQueriesBurst = 10
 )
 
 var (
@@ -251,6 +262,12 @@ type UniverseConfig struct {
 	FederationServers []string `long:"federationserver" description:"The host:port of a Universe server peer with. These servers will be added as the default set of federation servers. Can be specified multiple times."`
 
 	PublicAccess bool `long:"public-access" description:"If true, and the Universe server is on a public interface, valid proof from remote parties will be accepted, and proofs will be queryable by remote parties. This applies to federation syncing as well as RPC insert and query."`
+
+	StatsCacheDuration time.Duration `long:"stats-cache-duration" description:"The amount of time to cache stats for before refreshing them."`
+
+	UniverseQueriesPerSecond rate.Limit `long:"max-qps" description:"The maximum number of queries per second across the set of active universe queries that is permitted. Anything above this starts to get rate limited."`
+
+	UniverseQueriesBurst int `long:"req-burst-budget" description:"The burst budget for the universe query rate limiting."`
 }
 
 // AddressConfig is the config that houses any address Book related config
@@ -369,6 +386,10 @@ func DefaultConfig() Config {
 		},
 		Universe: &UniverseConfig{
 			SyncInterval: defaultUniverseSyncInterval,
+			UniverseQueriesPerSecond: rate.Limit(
+				defaultUniverseMaxQps,
+			),
+			UniverseQueriesBurst: defaultUniverseQueriesBurst,
 		},
 		AddrBook: &AddrBookConfig{
 			DisableSyncer: false,