diff --git a/cmd/flags/prover.go b/cmd/flags/prover.go index 6e1b19665..e7b6af69d 100644 --- a/cmd/flags/prover.go +++ b/cmd/flags/prover.go @@ -134,6 +134,12 @@ var ( Value: uint64(time.Hour.Seconds()), Category: proverCategory, } + TempCapacityExpiresAt = &cli.DurationFlag{ + Name: "prover.tempCapacityExpiresAt", + Usage: "time in seconds temporary capacity lives for, format: 36s", + Value: 36 * time.Second, + Category: proverCategory, + } ) // All prover flags. @@ -163,4 +169,5 @@ var ProverFlags = MergeFlags(CommonFlags, []cli.Flag{ ProverCapacity, MaxExpiry, TaikoTokenAddress, + TempCapacityExpiresAt, }) diff --git a/prover/capacity_manager/capacity_manager.go b/prover/capacity_manager/capacity_manager.go index 28daff6ff..1f0d0074d 100644 --- a/prover/capacity_manager/capacity_manager.go +++ b/prover/capacity_manager/capacity_manager.go @@ -2,20 +2,28 @@ package capacity_manager import ( "sync" + "time" "github.com/ethereum/go-ethereum/log" ) // CapacityManager manages the prover capacity concurrent-safely. type CapacityManager struct { - capacity uint64 - maxCapacity uint64 - mutex sync.RWMutex + capacity map[uint64]bool + tempCapacity []time.Time + tempCapacityExpiresAt time.Duration + maxCapacity uint64 + mutex sync.RWMutex } // New creates a new CapacityManager instance. -func New(capacity uint64) *CapacityManager { - return &CapacityManager{capacity: capacity, maxCapacity: capacity} +func New(capacity uint64, tempCapacityExpiresAt time.Duration) *CapacityManager { + return &CapacityManager{ + capacity: make(map[uint64]bool), + maxCapacity: capacity, + tempCapacity: make([]time.Time, 0), + tempCapacityExpiresAt: tempCapacityExpiresAt, + } } // ReadCapacity reads the current capacity. @@ -23,41 +31,76 @@ func (m *CapacityManager) ReadCapacity() uint64 { m.mutex.RLock() defer m.mutex.RUnlock() - log.Info("Reading capacity", "capacity", m.capacity) + log.Info("Reading capacity", "capacity", len(m.capacity)) - return m.capacity + return m.maxCapacity - uint64((len(m.capacity))) } // ReleaseOneCapacity releases one capacity. -func (m *CapacityManager) ReleaseOneCapacity() (uint64, bool) { +func (m *CapacityManager) ReleaseOneCapacity(blockID uint64) (uint64, bool) { m.mutex.Lock() defer m.mutex.Unlock() - if m.capacity+1 > m.maxCapacity { - log.Info("Can not release capacity", "currentCapacity", m.capacity, "maxCapacity", m.maxCapacity) - return m.capacity, false + if _, ok := m.capacity[blockID]; !ok { + log.Info("Can not release capacity", + "blockID", blockID, + "currentCapacity", m.maxCapacity-uint64(len(m.capacity)), + "maxCapacity", m.maxCapacity) + return uint64(len(m.capacity)), false } - m.capacity += 1 + delete(m.capacity, blockID) - log.Info("Released capacity", "capacityAfterRelease", m.capacity) + log.Info("Released capacity", "blockID", blockID, "capacityAfterRelease", len(m.capacity)) - return m.capacity, true + return m.maxCapacity - uint64(len(m.capacity)), true } // TakeOneCapacity takes one capacity. -func (m *CapacityManager) TakeOneCapacity() (uint64, bool) { +func (m *CapacityManager) TakeOneCapacity(blockID uint64) (uint64, bool) { m.mutex.Lock() defer m.mutex.Unlock() - if m.capacity == 0 { - log.Info("Could not take one capacity", "capacity", m.capacity) + if len(m.capacity) == int(m.maxCapacity) { + log.Info("Could not take one capacity", + "blockID", blockID, + "currentCapacity", m.maxCapacity-uint64(len(m.capacity))) return 0, false } - m.capacity -= 1 + m.capacity[blockID] = true + + log.Info("Took one capacity", + "blockID", blockID, + "capacityAfterTaking", m.maxCapacity-uint64(len(m.capacity))) + + return m.maxCapacity - uint64((len(m.capacity))), true +} + +func (m *CapacityManager) TakeOneTempCapacity() (uint64, bool) { + m.mutex.Lock() + defer m.mutex.Unlock() + + // clear expired tempCapacities - log.Info("Took one capacity", "capacityAfterTaking", m.capacity) + m.clearExpiredTempCapacities() - return m.capacity, true + if len(m.capacity)+len(m.tempCapacity) >= int(m.maxCapacity) { + log.Info("Could not take one temp capacity", + "capacity", m.maxCapacity-uint64(len(m.capacity)), + "tempCapacity", len(m.tempCapacity)) + return 0, false + } + + m.tempCapacity = append(m.tempCapacity, time.Now().UTC()) + + return m.maxCapacity - uint64(len(m.capacity)) - uint64((len(m.tempCapacity))), true +} + +func (m *CapacityManager) clearExpiredTempCapacities() { + for i, c := range m.tempCapacity { + if time.Now().UTC().Sub(c) > m.tempCapacityExpiresAt { + m.tempCapacity = append(m.tempCapacity[:i], m.tempCapacity[i+1:]...) + } + } } diff --git a/prover/capacity_manager/capacity_manager_test.go b/prover/capacity_manager/capacity_manager_test.go index be9ff36d1..1e7d7b4c4 100644 --- a/prover/capacity_manager/capacity_manager_test.go +++ b/prover/capacity_manager/capacity_manager_test.go @@ -1,11 +1,14 @@ package capacity_manager import ( + "time" + "github.com/stretchr/testify/suite" ) var ( - testCapacity uint64 = 1024 + testCapacity uint64 = 5 + tempCapacityExpiresAt time.Duration = 5 * time.Second ) type CapacityManagerTestSuite struct { @@ -14,7 +17,7 @@ type CapacityManagerTestSuite struct { } func (s *CapacityManagerTestSuite) SetupTest() { - s.m = New(testCapacity) + s.m = New(testCapacity, tempCapacityExpiresAt) } func (s *CapacityManagerTestSuite) TestReadCapacity() { @@ -22,7 +25,15 @@ func (s *CapacityManagerTestSuite) TestReadCapacity() { } func (s *CapacityManagerTestSuite) TestReleaseOneCapacity() { - capacity, released := s.m.ReleaseOneCapacity() + var blockID uint64 = 1 + _, released := s.m.ReleaseOneCapacity(blockID) + s.Equal(false, released) + + _, ok := s.m.TakeOneCapacity(blockID) + + s.Equal(true, ok) + + capacity, released := s.m.ReleaseOneCapacity(blockID) s.Equal(true, released) s.Equal(testCapacity+1, capacity) @@ -30,8 +41,56 @@ func (s *CapacityManagerTestSuite) TestReleaseOneCapacity() { } func (s *CapacityManagerTestSuite) TestTakeOneCapacity() { - capacity, ok := s.m.TakeOneCapacity() + var blockID uint64 = 1 + + capacity, ok := s.m.TakeOneCapacity(blockID) s.True(ok) s.Equal(testCapacity-1, capacity) s.Equal(testCapacity-1, s.m.ReadCapacity()) } + +func (s *CapacityManagerTestSuite) TestTakeOneTempCapacity() { + // take 3 actual capacity + var sl []uint64 = []uint64{1, 2, 3} + + for _, c := range sl { + _, ok := s.m.TakeOneCapacity(c) + s.True(ok) + } + + // should be 2 temp capacity left to take + capacity, ok := s.m.TakeOneTempCapacity() + s.True(ok) + s.Equal(int(testCapacity)-len(sl)-1, capacity) + + capacity, ok = s.m.TakeOneTempCapacity() + s.True(ok) + s.Equal(int(testCapacity)-len(sl)-2, capacity) + + // now it should fail, 3 capacity + 2 temp capacity + capacity, ok = s.m.TakeOneTempCapacity() + s.False(ok) + s.Equal(int(testCapacity)-len(sl)-2, capacity) + + // wait until they expire + time.Sleep(s.m.tempCapacityExpiresAt) + + // both should be expired, we should be able to take two more + capacity, ok = s.m.TakeOneTempCapacity() + s.True(ok) + s.Equal(int(testCapacity)-len(sl)-1, capacity) + + capacity, ok = s.m.TakeOneTempCapacity() + s.True(ok) + s.Equal(int(testCapacity)-len(sl)-2, capacity) + + // now remove one actual capacity, simulate "block done being proven" + capacity, ok = s.m.ReleaseOneCapacity(sl[0]) + s.True(ok) + s.Equal(int(testCapacity)-len(sl)-1, capacity) + + // and we should be able to take another temp capacity + capacity, ok = s.m.TakeOneTempCapacity() + s.True(ok) + s.Equal(int(testCapacity)-len(sl)-2, capacity) +} diff --git a/prover/config.go b/prover/config.go index b326c208b..c9c42d4a6 100644 --- a/prover/config.go +++ b/prover/config.go @@ -46,6 +46,7 @@ type Config struct { ProveBlockMaxTxGasTipCap *big.Int HTTPServerPort uint64 Capacity uint64 + TempCapacityExpiresAt time.Duration MinProofFee *big.Int MaxExpiry time.Duration } @@ -172,6 +173,7 @@ func NewConfigFromCliContext(c *cli.Context) (*Config, error) { WaitReceiptTimeout: time.Duration(c.Uint64(flags.WaitReceiptTimeout.Name)) * time.Second, ProveBlockGasLimit: proveBlockTxGasLimit, Capacity: c.Uint64(flags.ProverCapacity.Name), + TempCapacityExpiresAt: c.Duration(flags.TempCapacityExpiresAt.Name), ProveBlockTxReplacementMultiplier: proveBlockTxReplacementMultiplier, ProveBlockMaxTxGasTipCap: proveBlockMaxTxGasTipCap, HTTPServerPort: c.Uint64(flags.ProverHTTPServerPort.Name), diff --git a/prover/config_test.go b/prover/config_test.go index b671b2b39..130c507de 100644 --- a/prover/config_test.go +++ b/prover/config_test.go @@ -52,6 +52,7 @@ func (s *ProverTestSuite) TestNewConfigFromCliContext_OracleProver() { s.Equal(minProofFee, c.MinProofFee.String()) s.Equal(uint64(3), c.ProveBlockTxReplacementMultiplier) s.Equal(uint64(256), c.ProveBlockMaxTxGasTipCap.Uint64()) + s.Equal(15*time.Second, c.TempCapacityExpiresAt) s.Nil(new(Prover).InitFromCli(context.Background(), ctx)) s.True(c.ProveUnassignedBlocks) @@ -80,6 +81,7 @@ func (s *ProverTestSuite) TestNewConfigFromCliContext_OracleProver() { "--" + flags.OracleProverPrivateKey.Name, os.Getenv("L1_PROVER_PRIVATE_KEY"), "--" + flags.Graffiti.Name, "", "--" + flags.CheckProofWindowExpiredInterval.Name, "30", + "--" + flags.TempCapacityExpiresAt.Name, "15s", "--" + flags.ProveUnassignedBlocks.Name, })) } @@ -201,6 +203,7 @@ func (s *ProverTestSuite) SetupApp() *cli.App { &cli.Uint64Flag{Name: flags.ProverCapacity.Name}, &cli.Uint64Flag{Name: flags.MinProofFee.Name}, &cli.Uint64Flag{Name: flags.ProveBlockTxGasLimit.Name}, + &cli.DurationFlag{Name: flags.TempCapacityExpiresAt.Name}, } app.Action = func(ctx *cli.Context) error { _, err := NewConfigFromCliContext(ctx) diff --git a/prover/prover.go b/prover/prover.go index 7aa1ca2f2..4c9b2fe99 100644 --- a/prover/prover.go +++ b/prover/prover.go @@ -112,7 +112,7 @@ func InitFromConfig(ctx context.Context, p *Prover, cfg *Config) (err error) { p.currentBlocksBeingProvenMutex = new(sync.Mutex) p.currentBlocksWaitingForProofWindow = make(map[uint64]uint64, 0) p.currentBlocksWaitingForProofWindowMutex = new(sync.Mutex) - p.capacityManager = capacity.New(cfg.Capacity) + p.capacityManager = capacity.New(cfg.Capacity, cfg.TempCapacityExpiresAt) // Clients if p.rpc, err = rpc.NewClient(p.ctx, &rpc.ClientConfig{ @@ -604,7 +604,7 @@ func (p *Prover) onBlockProposed( } if !p.cfg.OracleProver { - if _, ok := p.capacityManager.TakeOneCapacity(); !ok { + if _, ok := p.capacityManager.TakeOneCapacity(event.BlockId.Uint64()); !ok { return errNoCapacity } } @@ -658,7 +658,7 @@ func (p *Prover) submitProofOp(ctx context.Context, proofWithHeader *proofProduc defer func() { <-p.submitProofConcurrencyGuard if !p.cfg.OracleProver { - _, released := p.capacityManager.ReleaseOneCapacity() + _, released := p.capacityManager.ReleaseOneCapacity(proofWithHeader.Meta.Id) if !released { log.Error("unable to release capacity") } @@ -876,9 +876,12 @@ func (p *Prover) cancelProof(ctx context.Context, blockID uint64) { cancel() delete(p.currentBlocksBeingProven, blockID) if !p.cfg.OracleProver { - capacity, released := p.capacityManager.ReleaseOneCapacity() + capacity, released := p.capacityManager.ReleaseOneCapacity(blockID) if !released { - log.Error("unable to release capacity while cancelling proof", "capacity", capacity) + log.Error("unable to release capacity while cancelling proof", + "capacity", capacity, + "blockID", blockID, + ) } } } @@ -1011,7 +1014,7 @@ func (p *Prover) requestProofForBlockId(blockId *big.Int, l1Height *big.Int) err // make sure to takea capacity before requesting proof if !p.cfg.OracleProver { - if _, ok := p.capacityManager.TakeOneCapacity(); !ok { + if _, ok := p.capacityManager.TakeOneCapacity(event.BlockId.Uint64()); !ok { return errNoCapacity } } diff --git a/prover/server/api.go b/prover/server/api.go index 921d81145..2d38f81f9 100644 --- a/prover/server/api.go +++ b/prover/server/api.go @@ -99,8 +99,8 @@ func (srv *ProverServer) CreateAssignment(c echo.Context) error { return echo.NewHTTPError(http.StatusUnprocessableEntity, "expiry too long") } - if srv.capacityManager.ReadCapacity() == 0 { - log.Warn("Prover does not have capacity", "proposerIP", c.RealIP()) + if _, ok := srv.capacityManager.TakeOneTempCapacity(); !ok { + log.Warn("Prover unable to take a temporary capacity", "proposerIP", c.RealIP()) return echo.NewHTTPError(http.StatusUnprocessableEntity, "prover does not have capacity") } diff --git a/prover/server/server_test.go b/prover/server/server_test.go index e8b5582e6..781ac3416 100644 --- a/prover/server/server_test.go +++ b/prover/server/server_test.go @@ -50,7 +50,7 @@ func (s *ProverServerTestSuite) SetupTest() { proverPrivateKey: l1ProverPrivKey, minProofFee: common.Big1, maxExpiry: 24 * time.Hour, - capacityManager: capacity.New(1024), + capacityManager: capacity.New(1024, 100*time.Second), taikoL1Address: common.HexToAddress(os.Getenv("TAIKO_L1_ADDRESS")), rpc: rpcClient, bond: common.Big0, diff --git a/testutils/suite.go b/testutils/suite.go index fc9713782..5ae1b09cd 100644 --- a/testutils/suite.go +++ b/testutils/suite.go @@ -6,6 +6,7 @@ import ( "math/big" "net/url" "os" + "time" "github.com/cenkalti/backoff/v4" "github.com/ethereum/go-ethereum/accounts/abi/bind" @@ -74,7 +75,7 @@ func (s *ClientTestSuite) SetupTest() { s.Nil(err) s.ProverEndpoints = []*url.URL{LocalRandomProverEndpoint()} - s.proverServer = NewTestProverServer(s, l1ProverPrivKey, capacity.New(1024), s.ProverEndpoints[0]) + s.proverServer = NewTestProverServer(s, l1ProverPrivKey, capacity.New(1024, 100*time.Second), s.ProverEndpoints[0]) tokenBalance, err := rpcCli.TaikoL1.GetTaikoTokenBalance(nil, crypto.PubkeyToAddress(l1ProverPrivKey.PublicKey)) s.Nil(err)