Merge pull request #253 from SiaFoundation/nate/max-rpcs

n8maninger · web-flow · commit 6e7a11a0549c · 2025-06-19T06:41:48.000-07:00
Increase max RPCs
diff --git a/.changeset/increase_max_inflight_rpcs_to_handle_mainnet_traffic.md b/.changeset/increase_max_inflight_rpcs_to_handle_mainnet_traffic.md
@@ -0,0 +1,5 @@
+---
+default: patch
+---
+
+# Increase max inflight RPCs to handle Mainnet traffic.
diff --git a/cmd/explored/main.go b/cmd/explored/main.go
@@ -256,7 +256,9 @@ func runRootCmd(ctx context.Context, log *zap.Logger) error {
 		UniqueID:   gateway.GenerateUniqueID(),
 		NetAddress: syncerAddr,
 	}
-	s := syncer.New(syncerListener, cm, ps, header, syncer.WithLogger(log.Named("syncer")), syncer.WithMaxInboundPeers(256))
+	s := syncer.New(syncerListener, cm, ps, header, syncer.WithLogger(log.Named("syncer")),
+		syncer.WithMaxInboundPeers(256),
+		syncer.WithMaxInflightRPCs(1024))
 	defer s.Close()
 	go s.Run()
 
diff --git a/explorer/explorer.go b/explorer/explorer.go
@@ -146,7 +146,6 @@ func (e *Explorer) syncStore(index types.ChainIndex, batchSize int) error {
 			if err != nil {
 				return fmt.Errorf("failed to subscribe to chain manager: %w", err)
 			}
-
 			if err := e.s.UpdateChainState(crus, caus); err != nil {
 				return fmt.Errorf("failed to process updates: %w", err)
 			}
diff --git a/explorer/update.go b/explorer/update.go
@@ -5,6 +5,7 @@ import (
 
 	"go.sia.tech/core/types"
 	"go.sia.tech/coreutils/chain"
+	"go.uber.org/zap"
 )
 
 type (
@@ -589,21 +590,25 @@ func updateMetrics(tx UpdateTx, s UpdateState, metrics Metrics) (Metrics, error)
 }
 
 // UpdateChainState applies the reverts and updates.
-func UpdateChainState(tx UpdateTx, crus []chain.RevertUpdate, caus []chain.ApplyUpdate) error {
+func UpdateChainState(tx UpdateTx, crus []chain.RevertUpdate, caus []chain.ApplyUpdate, log *zap.Logger) error {
 	for _, cru := range crus {
 		revertedIndex := types.ChainIndex{
 			ID:     cru.Block.ID(),
 			Height: cru.State.Index.Height + 1,
 		}
+		log.Debug("reverting chain update", zap.Stringer("index", revertedIndex))
 		if err := revertChainUpdate(tx, cru, revertedIndex); err != nil {
 			return fmt.Errorf("failed to revert chain update %q: %w", revertedIndex, err)
 		}
+		log.Debug("reverted chain update", zap.Stringer("index", revertedIndex))
 	}
 
 	for _, cau := range caus {
+		log.Debug("applying chain update", zap.Stringer("index", cau.State.Index))
 		if err := applyChainUpdate(tx, cau); err != nil {
 			return fmt.Errorf("failed to apply chain update %q: %w", cau.State.Index, err)
 		}
+		log.Debug("applied chain update", zap.Stringer("index", cau.State.Index))
 	}
 	return nil
 }
diff --git a/go.mod b/go.mod
@@ -6,8 +6,8 @@ require (
 	github.com/google/go-cmp v0.7.0
 	github.com/mattn/go-sqlite3 v1.14.28
 	github.com/oschwald/geoip2-golang v1.11.0
-	go.sia.tech/core v0.13.2
-	go.sia.tech/coreutils v0.16.2
+	go.sia.tech/core v0.13.3-0.20250616154238-4c58987023c7
+	go.sia.tech/coreutils v0.16.3-0.20250619115925-ab2381254465
 	go.sia.tech/jape v0.14.0
 	go.uber.org/zap v1.27.0
 	golang.org/x/exp v0.0.0-20240506185415-9bf2ced13842
diff --git a/go.sum b/go.sum
@@ -43,18 +43,12 @@ github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+
 github.com/stretchr/testify v1.6.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
 github.com/stretchr/testify v1.10.0 h1:Xv5erBjTwe/5IxqUQTdXv5kgmIvbHo3QQyRwhJsOfJA=
 github.com/stretchr/testify v1.10.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
-go.etcd.io/bbolt v1.4.0 h1:TU77id3TnN/zKr7CO/uk+fBCwF2jGcMuw2B/FMAzYIk=
-go.etcd.io/bbolt v1.4.0/go.mod h1:AsD+OCi/qPN1giOX1aiLAha3o1U8rAz65bvN4j0sRuk=
 go.etcd.io/bbolt v1.4.1 h1:5mOV+HWjIPLEAlUGMsveaUvK2+byZMFOzojoi7bh7uI=
 go.etcd.io/bbolt v1.4.1/go.mod h1:c8zu2BnXWTu2XM4XcICtbGSl9cFwsXtcf9zLt2OncM8=
-go.sia.tech/core v0.13.2 h1:66ZYzN2+AiHZmRayt4idSfoEoQZ5LV541+2E0luiizA=
-go.sia.tech/core v0.13.2/go.mod h1:bur1jeLA1JQbwzZkc2ijSTdpJYusG4h0pV9IwLOHT0g=
-go.sia.tech/coreutils v0.16.2-0.20250614154532-fc51637947af h1:s32mW2KWeE4/HTTE2h9GUqVmr3bH8rTA9Sa7iTBlWrk=
-go.sia.tech/coreutils v0.16.2-0.20250614154532-fc51637947af/go.mod h1:BPoX9f3/ViDi6STLMoeziZZKCJarVwGV3yXv5pl0opQ=
-go.sia.tech/coreutils v0.16.2-0.20250616161340-e162bc863b6d h1:2WxA3RDJltSyfSLmpaNmD02BQ928zWLouvWVTwj+jCA=
-go.sia.tech/coreutils v0.16.2-0.20250616161340-e162bc863b6d/go.mod h1:EWSwOvs9POy172NRn5CwgYVUH5TENcAODKaEanXgkmU=
-go.sia.tech/coreutils v0.16.2 h1:sAVwl7s9bBqBp9q1C93J9aY6ijdEtYHIhZkrde2GLsE=
-go.sia.tech/coreutils v0.16.2/go.mod h1:egAAlR7vju4nMppZiIEfHcwr84abuW9wg/ytcu4kSnA=
+go.sia.tech/core v0.13.3-0.20250616154238-4c58987023c7 h1:4KCpwSuMMZuiWcSOpQ9GFLpGhIxVdaqOnd4Bz2U590c=
+go.sia.tech/core v0.13.3-0.20250616154238-4c58987023c7/go.mod h1:bur1jeLA1JQbwzZkc2ijSTdpJYusG4h0pV9IwLOHT0g=
+go.sia.tech/coreutils v0.16.3-0.20250619115925-ab2381254465 h1:WrsOPXNWSNTu7EJ2URm1utTyes0Yc9779faYWRWm1lU=
+go.sia.tech/coreutils v0.16.3-0.20250619115925-ab2381254465/go.mod h1:tcO75XD7wSV+neJJiS036882NnWSfOXDcU9nlw1r6U8=
 go.sia.tech/jape v0.14.0 h1:hyocTKqvcji+rC1vDE1djINlpErQQVDS6zoLMmxW3Xs=
 go.sia.tech/jape v0.14.0/go.mod h1:tONxoKrNr0iQWzBCygwlTkGoGjuEhyVpLGInvGd2mGY=
 go.sia.tech/mux v1.4.0 h1:LgsLHtn7l+25MwrgaPaUCaS8f2W2/tfvHIdXps04sVo=
diff --git a/persist/sqlite/consensus.go b/persist/sqlite/consensus.go
@@ -1167,7 +1167,7 @@ func (s *Store) UpdateChainState(reverted []chain.RevertUpdate, applied []chain.
 			tx: tx,
 		}
 
-		if err := explorer.UpdateChainState(utx, reverted, applied); err != nil {
+		if err := explorer.UpdateChainState(utx, reverted, applied, s.log.Named("UpdateChainState")); err != nil {
 			return fmt.Errorf("failed to update chain state: %w", err)
 		}
 		return nil
diff --git a/persist/sqlite/consensus_refactored_test.go b/persist/sqlite/consensus_refactored_test.go
@@ -3141,7 +3141,7 @@ func BenchmarkRevert(b *testing.B) {
 			}
 
 			b.StartTimer()
-			err := explorer.UpdateChainState(utx, crus, nil)
+			err := explorer.UpdateChainState(utx, crus, nil, n.db.log.Named("update"))
 			b.StopTimer()
 
 			if err != nil {

-Original file line number
+Diff line change
@@ @@ -0,0 +1,5 @@ @@
 +---
 +default: patch
 +---
++
 +# Increase max inflight RPCs to handle Mainnet traffic.
Original file line number	Diff line number	Diff line change
`@@ -146,7 +146,6 @@ func (e *Explorer) syncStore(index types.ChainIndex, batchSize int) error {`
`146`	`146`	`if err != nil {`
`147`	`147`	`return fmt.Errorf("failed to subscribe to chain manager: %w", err)`
`148`	`148`	`}`
`149`		`-`
`150`	`149`	`if err := e.s.UpdateChainState(crus, caus); err != nil {`
`151`	`150`	`return fmt.Errorf("failed to process updates: %w", err)`
`152`	`151`	`}`
Original file line number	Diff line number	Diff line change
`@@ -5,6 +5,7 @@ import (`
`5`	`5`
`6`	`6`	`"go.sia.tech/core/types"`
`7`	`7`	`"go.sia.tech/coreutils/chain"`
	`8`	`+ "go.uber.org/zap"`
`8`	`9`	`)`
`9`	`10`
`10`	`11`	`type (`
`@@ -589,21 +590,25 @@ func updateMetrics(tx UpdateTx, s UpdateState, metrics Metrics) (Metrics, error)`
`589`	`590`	`}`
`590`	`591`
`591`	`592`	`// UpdateChainState applies the reverts and updates.`
`592`		`-func UpdateChainState(tx UpdateTx, crus []chain.RevertUpdate, caus []chain.ApplyUpdate) error {`
	`593`	`+func UpdateChainState(tx UpdateTx, crus []chain.RevertUpdate, caus []chain.ApplyUpdate, log *zap.Logger) error {`
`593`	`594`	`for _, cru := range crus {`
`594`	`595`	`revertedIndex := types.ChainIndex{`
`595`	`596`	`ID: cru.Block.ID(),`
`596`	`597`	`Height: cru.State.Index.Height + 1,`
`597`	`598`	`}`
	`599`	`+ log.Debug("reverting chain update", zap.Stringer("index", revertedIndex))`
`598`	`600`	`if err := revertChainUpdate(tx, cru, revertedIndex); err != nil {`
`599`	`601`	`return fmt.Errorf("failed to revert chain update %q: %w", revertedIndex, err)`
`600`	`602`	`}`
	`603`	`+ log.Debug("reverted chain update", zap.Stringer("index", revertedIndex))`
`601`	`604`	`}`
`602`	`605`
`603`	`606`	`for _, cau := range caus {`
	`607`	`+ log.Debug("applying chain update", zap.Stringer("index", cau.State.Index))`
`604`	`608`	`if err := applyChainUpdate(tx, cau); err != nil {`
`605`	`609`	`return fmt.Errorf("failed to apply chain update %q: %w", cau.State.Index, err)`
`606`	`610`	`}`
	`611`	`+ log.Debug("applied chain update", zap.Stringer("index", cau.State.Index))`
`607`	`612`	`}`
`608`	`613`	`return nil`
`609`	`614`	`}`
Original file line number	Diff line number	Diff line change
`@@ -1167,7 +1167,7 @@ func (s *Store) UpdateChainState(reverted []chain.RevertUpdate, applied []chain.`
`1167`	`1167`	`tx: tx,`
`1168`	`1168`	`}`
`1169`	`1169`
`1170`		`- if err := explorer.UpdateChainState(utx, reverted, applied); err != nil {`
	`1170`	`+ if err := explorer.UpdateChainState(utx, reverted, applied, s.log.Named("UpdateChainState")); err != nil {`
`1171`	`1171`	`return fmt.Errorf("failed to update chain state: %w", err)`
`1172`	`1172`	`}`
`1173`	`1173`	`return nil`
Original file line number	Diff line number	Diff line change
`@@ -3141,7 +3141,7 @@ func BenchmarkRevert(b *testing.B) {`
`3141`	`3141`	`}`
`3142`	`3142`
`3143`	`3143`	`b.StartTimer()`
`3144`		`- err := explorer.UpdateChainState(utx, crus, nil)`
	`3144`	`+ err := explorer.UpdateChainState(utx, crus, nil, n.db.log.Named("update"))`
`3145`	`3145`	`b.StopTimer()`
`3146`	`3146`
`3147`	`3147`	`if err != nil {`