feat: withdrawalQueue and deallocationQueue logic into rewards calculation

cmd/database.go

@@ -114,7 +114,7 @@ var runDatabaseCmd = &cobra.Command{
 			l.Sugar().Fatalw("Failed to load meta state models", zap.Error(err))
 		}
 
-		precommitProcessors.LoadPrecommitProcessors(sm, grm, l)
+		precommitProcessors.LoadPrecommitProcessors(sm, grm, l, cfg)
 
 		fetchr := fetcher.NewFetcher(client, &fetcher.FetcherConfig{UseGetBlockReceipts: cfg.EthereumRpcConfig.UseGetBlockReceipts}, contractStore, l)
 

cmd/debugger.go

@@ -116,7 +116,7 @@ var runDebuggerCmd = &cobra.Command{
 			l.Sugar().Fatalw("Failed to load meta state models", zap.Error(err))
 		}
 
-		precommitProcessors.LoadPrecommitProcessors(sm, grm, l)
+		precommitProcessors.LoadPrecommitProcessors(sm, grm, l, cfg)
 
 		fetchr := fetcher.NewFetcher(client, &fetcher.FetcherConfig{UseGetBlockReceipts: cfg.EthereumRpcConfig.UseGetBlockReceipts}, contractStore, l)
 

cmd/debugger/main.go

@@ -114,7 +114,7 @@ func main() {
 		l.Sugar().Fatalw("Failed to load meta state models", zap.Error(err))
 	}
 
-	precommitProcessors.LoadPrecommitProcessors(sm, grm, l)
+	precommitProcessors.LoadPrecommitProcessors(sm, grm, l, cfg)
 
 	fetchr := fetcher.NewFetcher(client, &fetcher.FetcherConfig{UseGetBlockReceipts: cfg.EthereumRpcConfig.UseGetBlockReceipts}, contractStore, l)
 

cmd/operatorRestakedStrategies.go

@@ -86,7 +86,7 @@ var runOperatorRestakedStrategiesCmd = &cobra.Command{
 			l.Sugar().Fatalw("Failed to load eigen state models", zap.Error(err))
 		}
 
-		precommitProcessors.LoadPrecommitProcessors(sm, grm, l)
+		precommitProcessors.LoadPrecommitProcessors(sm, grm, l, cfg)
 
 		fetchr := fetcher.NewFetcher(client, &fetcher.FetcherConfig{UseGetBlockReceipts: cfg.EthereumRpcConfig.UseGetBlockReceipts}, contractStore, l)
 

cmd/rpc.go

@@ -119,7 +119,7 @@ var rpcCmd = &cobra.Command{
 			l.Sugar().Fatalw("Failed to load eigen state models", zap.Error(err))
 		}
 
-		precommitProcessors.LoadPrecommitProcessors(sm, grm, l)
+		precommitProcessors.LoadPrecommitProcessors(sm, grm, l, cfg)
 
 		sog := stakerOperators.NewStakerOperatorGenerator(grm, l, cfg)
 

cmd/run.go

@@ -150,7 +150,7 @@ var runCmd = &cobra.Command{
 			l.Sugar().Fatalw("Failed to load meta state models", zap.Error(err))
 		}
 
-		precommitProcessors.LoadPrecommitProcessors(sm, grm, l)
+		precommitProcessors.LoadPrecommitProcessors(sm, grm, l, cfg)
 
 		fetchr := fetcher.NewFetcher(client, &fetcher.FetcherConfig{UseGetBlockReceipts: cfg.EthereumRpcConfig.UseGetBlockReceipts}, contractStore, l)
 

internal/config/config.go

@@ -54,6 +54,18 @@ func normalizeFlagName(name string) string {
 	return strings.ReplaceAll(name, "-", "_")
 }
 
+// getDefaultWithdrawalQueueDuration returns the default withdrawal queue duration in days based on chain
+func getDefaultWithdrawalQueueDuration(chain Chain) float64 {
+	switch chain {
+	case Chain_Mainnet:
+		return 14.0 // Mainnet uses 14-day withdrawal queue
+	case Chain_Preprod, Chain_Holesky, Chain_Sepolia, Chain_Hoodi, Chain_PreprodHoodi:
+		return 10.0 / (24.0 * 60.0) // Testnet/preprod uses 10 minutes = ~0.0069 days
+	default:
+		return 14.0 // Default to mainnet behavior
+	}
+}
+
 type EthereumRpcConfig struct {
 	BaseUrl               string
 	ContractCallBatchSize int  // Number of contract calls to make in parallel
@@ -105,6 +117,7 @@ type RewardsConfig struct {
 	ValidateRewardsRoot          bool
 	GenerateStakerOperatorsTable bool
 	CalculateRewardsDaily        bool
+	WithdrawalQueueWindow        float64 // Duration in days for withdrawal queue period (14.0 for mainnet, 0.0069 for testnet/preprod ~10 min)
 }
 
 type StatsdConfig struct {
@@ -185,6 +198,20 @@ func StringWithDefaults(values ...string) string {
 	return ""
 }
 
+func IntWithDefault(value, defaultValue int) int {
+	if value == 0 {
+		return defaultValue
+	}
+	return value
+}
+
+func FloatWithDefault(value, defaultValue float64) float64 {
+	if value == 0.0 {
+		return defaultValue
+	}
+	return value
+}
+
 var (
 	Debug               = "debug"
 	DatabaseHost        = "database.host"
@@ -213,6 +240,7 @@ var (
 	RewardsValidateRewardsRoot          = "rewards.validate_rewards_root"
 	RewardsGenerateStakerOperatorsTable = "rewards.generate_staker_operators_table"
 	RewardsCalculateRewardsDaily        = "rewards.calculate_rewards_daily"
+	RewardsWithdrawalQueueWindow        = "rewards.withdrawal_queue_window"
 
 	EthereumRpcBaseUrl               = "ethereum.rpc_url"
 	EthereumRpcContractCallBatchSize = "ethereum.contract_call_batch_size"
@@ -248,9 +276,11 @@ var (
 )
 
 func NewConfig() *Config {
+	chain := Chain(StringWithDefault(viper.GetString(normalizeFlagName("chain")), "holesky"))
+
 	return &Config{
 		Debug: viper.GetBool(normalizeFlagName("debug")),
-		Chain: Chain(StringWithDefault(viper.GetString(normalizeFlagName("chain")), "holesky")),
+		Chain: chain,
 
 		EthereumRpcConfig: EthereumRpcConfig{
 			BaseUrl:               viper.GetString(normalizeFlagName(EthereumRpcBaseUrl)),
@@ -298,6 +328,7 @@ func NewConfig() *Config {
 			ValidateRewardsRoot:          viper.GetBool(normalizeFlagName(RewardsValidateRewardsRoot)),
 			GenerateStakerOperatorsTable: viper.GetBool(normalizeFlagName(RewardsGenerateStakerOperatorsTable)),
 			CalculateRewardsDaily:        viper.GetBool(normalizeFlagName(RewardsCalculateRewardsDaily)),
+			WithdrawalQueueWindow:        FloatWithDefault(viper.GetFloat64(normalizeFlagName(RewardsWithdrawalQueueWindow)), getDefaultWithdrawalQueueDuration(chain)),
 		},
 
 		DataDogConfig: DataDogConfig{

pkg/eigenState/precommitProcessors/precommitProcessors.go

@@ -1,12 +1,13 @@
 package precommitProcessors
 
 import (
+	"github.com/Layr-Labs/sidecar/internal/config"
 	"github.com/Layr-Labs/sidecar/pkg/eigenState/precommitProcessors/slashingProcessor"
 	"github.com/Layr-Labs/sidecar/pkg/eigenState/stateManager"
 	"go.uber.org/zap"
 	"gorm.io/gorm"
 )
 
-func LoadPrecommitProcessors(sm *stateManager.EigenStateManager, grm *gorm.DB, l *zap.Logger) {
-	slashingProcessor.NewSlashingProcessor(sm, l, grm)
+func LoadPrecommitProcessors(sm *stateManager.EigenStateManager, grm *gorm.DB, l *zap.Logger, cfg *config.Config) {
+	slashingProcessor.NewSlashingProcessor(sm, l, grm, cfg)
 }

pkg/eigenState/precommitProcessors/slashingProcessor/slashing.go

@@ -2,6 +2,7 @@ package slashingProcessor
 
 import (
 	"fmt"
+	"github.com/Layr-Labs/sidecar/internal/config"
 	"github.com/Layr-Labs/sidecar/pkg/eigenState/stakerDelegations"
 	"github.com/Layr-Labs/sidecar/pkg/eigenState/stakerShares"
 	"github.com/Layr-Labs/sidecar/pkg/eigenState/stateManager"
@@ -11,14 +12,16 @@ import (
 )
 
 type SlashingProcessor struct {
-	logger *zap.Logger
-	grm    *gorm.DB
+	logger       *zap.Logger
+	grm          *gorm.DB
+	globalConfig *config.Config
 }
 
-func NewSlashingProcessor(sm *stateManager.EigenStateManager, logger *zap.Logger, grm *gorm.DB) *SlashingProcessor {
+func NewSlashingProcessor(sm *stateManager.EigenStateManager, logger *zap.Logger, grm *gorm.DB, cfg *config.Config) *SlashingProcessor {
 	processor := &SlashingProcessor{
-		logger: logger,
-		grm:    grm,
+		logger:       logger,
+		grm:          grm,
+		globalConfig: cfg,
 	}
 	sm.RegisterPrecommitProcessor(processor, 0)
 	return processor
@@ -80,5 +83,173 @@ func (sp *SlashingProcessor) Process(blockNumber uint64, models map[string]types
 		precommitDelegations = append(precommitDelegations, delegation)
 	}
 	stakerSharesModel.PrecommitDelegatedStakers[blockNumber] = precommitDelegations
+
+	// Also handle slashing adjustments for queued withdrawals
+	err := sp.processQueuedWithdrawalSlashing(blockNumber, models)
+	if err != nil {
+		sp.logger.Sugar().Errorw("Failed to process queued withdrawal slashing",
+			zap.Error(err),
+			zap.Uint64("blockNumber", blockNumber),
+		)
+		return err
+	}
+
+	return nil
+}
+
+// SlashingEvent represents a slashing event from the database
+type SlashingEvent struct {
+	Operator        string
+	Strategy        string
+	WadSlashed      string
+	TransactionHash string
+	LogIndex        uint64
+}
+
+// processQueuedWithdrawalSlashing creates adjustment records for queued withdrawals
+// when an operator is slashed, so that the effective withdrawal amount is reduced.
+func (sp *SlashingProcessor) processQueuedWithdrawalSlashing(blockNumber uint64, models map[string]types.IEigenStateModel) error {
+	// Query slashed_operator_shares table directly for this block's slashing events
+	var slashingEvents []SlashingEvent
+	err := sp.grm.Table("slashed_operator_shares").
+		Where("block_number = ?", blockNumber).
+		Find(&slashingEvents).Error
+
+	if err != nil {
+		sp.logger.Sugar().Errorw("Failed to query slashing events",
+			zap.Error(err),
+			zap.Uint64("blockNumber", blockNumber),
+		)
+		return err
+	}
+
+	if len(slashingEvents) == 0 {
+		sp.logger.Sugar().Debug("No slashing events found for block number", zap.Uint64("blockNumber", blockNumber))
+		return nil
+	}
+
+	// For each slashing event, find active queued withdrawals and create adjustment records
+	for i := range slashingEvents {
+		slashEvent := &slashingEvents[i]
+		err := sp.createSlashingAdjustments(slashEvent, blockNumber)
+		if err != nil {
+			sp.logger.Sugar().Errorw("Failed to create slashing adjustments",
+				zap.Error(err),
+				zap.Uint64("blockNumber", blockNumber),
+				zap.String("operator", slashEvent.Operator),
+				zap.String("strategy", slashEvent.Strategy),
+			)
+			return err
+		}
+	}
+
+	return nil
+}
+
+func (sp *SlashingProcessor) createSlashingAdjustments(slashEvent *SlashingEvent, blockNumber uint64) error {
+	// Find all active queued withdrawals for this operator/strategy
+	query := `
+		SELECT
+			qsw.staker,
+			qsw.strategy,
+			qsw.operator,
+			qsw.block_number as withdrawal_block_number,
+			@slashBlockNumber as slash_block_number,
+			-- Calculate cumulative slash multiplier: previous multipliers * (1 - current_slash)
+			COALESCE(
+				(SELECT slash_multiplier
+				 FROM queued_withdrawal_slashing_adjustments adj
+				 WHERE adj.staker = qsw.staker
+				 AND adj.strategy = qsw.strategy
+				 AND adj.operator = qsw.operator
+				 AND adj.withdrawal_block_number = qsw.block_number
+				 ORDER BY adj.slash_block_number DESC
+				 LIMIT 1),
+				1
+			) * (1 - LEAST(@wadSlashed / 1e18, 1)) as slash_multiplier,
+			@blockNumber as block_number,
+			@transactionHash as transaction_hash,
+			@logIndex as log_index
+		FROM queued_slashing_withdrawals qsw
+		INNER JOIN blocks b_queued ON qsw.block_number = b_queued.number
+		WHERE qsw.operator = @operator
+		AND qsw.strategy = @strategy
+		-- Withdrawal was queued before this slash (check block number AND log index for same-block events)
+		AND (
+			qsw.block_number < @slashBlockNumber
+			OR (qsw.block_number = @slashBlockNumber AND qsw.log_index < @logIndex)
+		)
+		-- Still within withdrawal queue window (not yet completable)
+		AND b_queued.block_time + (@withdrawalQueueWindow * INTERVAL '1 day') > (
+			SELECT block_time FROM blocks WHERE number = @blockNumber
+		)
+		-- Backwards compatibility: only process records with valid data
+		AND qsw.staker IS NOT NULL
+		AND qsw.strategy IS NOT NULL
+		AND qsw.operator IS NOT NULL
+		AND qsw.shares_to_withdraw IS NOT NULL
+		AND b_queued.block_time IS NOT NULL
+	`
+
+	type AdjustmentRecord struct {
+		Staker                string
+		Strategy              string
+		Operator              string
+		WithdrawalBlockNumber uint64
+		SlashBlockNumber      uint64
+		SlashMultiplier       string
+		BlockNumber           uint64
+		TransactionHash       string
+		LogIndex              uint64
+	}
+
+	var adjustments []AdjustmentRecord
+	err := sp.grm.Raw(query, map[string]any{
+		"slashBlockNumber":      blockNumber,
+		"wadSlashed":            slashEvent.WadSlashed,
+		"blockNumber":           blockNumber,
+		"transactionHash":       slashEvent.TransactionHash,
+		"logIndex":              slashEvent.LogIndex,
+		"operator":              slashEvent.Operator,
+		"strategy":              slashEvent.Strategy,
+		"withdrawalQueueWindow": sp.globalConfig.Rewards.WithdrawalQueueWindow,
+	}).Scan(&adjustments).Error
+
+	if err != nil {
+		return fmt.Errorf("failed to find active withdrawals for slashing: %w", err)
+	}
+
+	if len(adjustments) == 0 {
+		sp.logger.Sugar().Debugw("No active queued withdrawals found for slashing event",
+			zap.String("operator", slashEvent.Operator),
+			zap.String("strategy", slashEvent.Strategy),
+			zap.Uint64("blockNumber", blockNumber),
+		)
+		return nil
+	}
+
+	// Insert adjustment records
+	for _, adj := range adjustments {
+		err := sp.grm.Table("queued_withdrawal_slashing_adjustments").Create(&adj).Error
+		if err != nil {
+			sp.logger.Sugar().Errorw("Failed to create slashing adjustment record",
+				zap.Error(err),
+				zap.String("staker", adj.Staker),
+				zap.String("strategy", adj.Strategy),
+				zap.Uint64("withdrawalBlockNumber", adj.WithdrawalBlockNumber),
+			)
+			return err
+		}
+
+		sp.logger.Sugar().Infow("Created queued withdrawal slashing adjustment",
+			zap.String("staker", adj.Staker),
+			zap.String("strategy", adj.Strategy),
+			zap.String("operator", adj.Operator),
+			zap.Uint64("withdrawalBlock", adj.WithdrawalBlockNumber),
+			zap.Uint64("slashBlock", adj.SlashBlockNumber),
+			zap.String("multiplier", adj.SlashMultiplier),
+		)
+	}
+
 	return nil
 }

pkg/eigenState/precommitProcessors/slashingProcessor/slashing_test.go

@@ -41,8 +41,8 @@ func setup() (
 	return dbname, grm, l, cfg, nil
 }
 
-func withSlashingProcessor(esm *stateManager.EigenStateManager, grm *gorm.DB, l *zap.Logger) *SlashingProcessor {
-	return NewSlashingProcessor(esm, l, grm)
+func withSlashingProcessor(esm *stateManager.EigenStateManager, grm *gorm.DB, l *zap.Logger, cfg *config.Config) *SlashingProcessor {
+	return NewSlashingProcessor(esm, l, grm, cfg)
 }
 
 func teardown(db *gorm.DB) {
@@ -67,7 +67,7 @@ func Test_SlashingPrecommitProcessor(t *testing.T) {
 
 	t.Run("Should capture delegate, deposit, slash in same block", func(t *testing.T) {
 		esm := stateManager.NewEigenStateManager(nil, l, grm)
-		withSlashingProcessor(esm, grm, l)
+		withSlashingProcessor(esm, grm, l, cfg)
 
 		blockNumber := uint64(200)
 		err = createBlock(grm, blockNumber)
@@ -140,7 +140,7 @@ func Test_SlashingPrecommitProcessor(t *testing.T) {
 
 	t.Run("Should capture many deposits and slash in same block", func(t *testing.T) {
 		esm := stateManager.NewEigenStateManager(nil, l, grm)
-		withSlashingProcessor(esm, grm, l)
+		withSlashingProcessor(esm, grm, l, cfg)
 
 		blockNumber := uint64(200)
 		err = createBlock(grm, blockNumber)

pkg/eigenState/stateMigrator/stateMigrations/202502031222_operatorSets/migration.go

@@ -48,7 +48,7 @@ func (sm *StateMigration) MigrateState(currentBlockNumber uint64) ([][]byte, map
 		return nil, nil, err
 	}
 
-	precommitProcessors.LoadPrecommitProcessors(stateMan, sm.db, sm.logger)
+	precommitProcessors.LoadPrecommitProcessors(stateMan, sm.db, sm.logger, sm.globalConfig)
 
 	contracts := sm.globalConfig.GetContractsMapForChain()
 

pkg/eigenState/stateMigrator/stateMigrations/202503040918_slashingEvents/migration.go

@@ -47,7 +47,7 @@ func (sm *StateMigration) MigrateState(currentBlockNumber uint64) ([][]byte, map
 		return nil, nil, err
 	}
 
-	precommitProcessors.LoadPrecommitProcessors(stateMan, sm.db, sm.logger)
+	precommitProcessors.LoadPrecommitProcessors(stateMan, sm.db, sm.logger, sm.globalConfig)
 
 	contracts := sm.globalConfig.GetContractsMapForChain()