diff --git a/node/pkg/watchers/solana/client.go b/node/pkg/watchers/solana/client.go index 4848fe2912..c2c11c74d5 100644 --- a/node/pkg/watchers/solana/client.go +++ b/node/pkg/watchers/solana/client.go @@ -741,7 +741,14 @@ func (s *SolanaWatcher) processInstruction(ctx context.Context, logger *zap.Logg return false, fmt.Errorf("failed to determine commitment: %w", err) } - if level != s.commitment { + if !s.checkCommitment(level, isReobservation) { + if logger.Level().Enabled(zapcore.DebugLevel) { + logger.Debug("skipping message which does not match the watcher commitment", + zap.Stringer("signature", tx.Signatures[0]), + zap.String("message commitment", string(level)), + zap.String("watcher commitment", string(s.commitment)), + ) + } return true, nil } @@ -923,19 +930,16 @@ func (s *SolanaWatcher) processMessageAccount(logger *zap.Logger, data []byte, a zap.Error(err)) return } - if commitment != s.commitment { - if isReobservation && s.commitment == rpc.CommitmentFinalized { - // There is only a single reobservation request channel for each chain, which is assigned to the finalized watcher. - // If someone requests reobservation of a confirmed message, we should allow the observation to go through. - logger.Info("allowing reobservation although the commitment level does not match the watcher", - zap.Stringer("account", acc), zap.String("message commitment", string(commitment)), zap.String("watcher commitment", string(s.commitment)), + + if !s.checkCommitment(commitment, isReobservation) { + if logger.Level().Enabled(zapcore.DebugLevel) { + logger.Debug("skipping message which does not match the watcher commitment", + zap.Stringer("account", acc), + zap.String("message commitment", string(commitment)), + zap.String("watcher commitment", string(s.commitment)), ) - } else { - if logger.Level().Enabled(zapcore.DebugLevel) { - logger.Debug("skipping message which does not match the watcher commitment", zap.Stringer("account", acc), zap.String("message commitment", string(commitment)), zap.String("watcher commitment", string(s.commitment))) - } - return } + return } // As of 2023-11-09, Pythnet has a bug which is not zeroing out these fields appropriately. This carve out should be removed after a fix is deployed. @@ -1060,3 +1064,16 @@ func (s *SolanaWatcher) populateLookupTableAccounts(ctx context.Context, tx *sol return nil } + +// checkCommitment checks to see if the commitment level of an observation matches the watcher. If it does, the observation should be published. +// If the commitment level does not match and the message is not a reobservation, then it should be dropped. In the case of a reobservation +// where the commitment level doesn't match, we need to check to see if this is the finalized watcher. If it is, then we should generate the +// observation. This is because all reobservation requests are handled by the finalized watcher. +func (s *SolanaWatcher) checkCommitment(commitment rpc.CommitmentType, isReobservation bool) bool { + if commitment != s.commitment { + if !isReobservation || s.commitment != rpc.CommitmentFinalized { + return false + } + } + return true +} diff --git a/node/pkg/watchers/solana/client_test.go b/node/pkg/watchers/solana/client_test.go index baf1af2400..b350634c3c 100644 --- a/node/pkg/watchers/solana/client_test.go +++ b/node/pkg/watchers/solana/client_test.go @@ -6,6 +6,7 @@ import ( "github.com/stretchr/testify/assert" "github.com/gagliardetto/solana-go" + "github.com/gagliardetto/solana-go/rpc" ) func TestVerifyConstants(t *testing.T) { @@ -13,3 +14,45 @@ func TestVerifyConstants(t *testing.T) { assert.Equal(t, SolanaAccountLen, solana.PublicKeyLength) assert.Equal(t, SolanaSignatureLen, len(solana.Signature{})) } + +func TestCheckCommitment(t *testing.T) { + type test struct { + commitment string + watcher string + isReobservation bool + result bool + } + tests := []test{ + // New observation success cases + {commitment: "finalized", watcher: "finalized", isReobservation: false, result: true}, + {commitment: "confirmed", watcher: "confirmed", isReobservation: false, result: true}, + + // New observation failure cases + {commitment: "finalized", watcher: "confirmed", isReobservation: false, result: false}, + {commitment: "confirmed", watcher: "finalized", isReobservation: false, result: false}, + + // Reobservation success cases + {commitment: "finalized", watcher: "finalized", isReobservation: true, result: true}, + {commitment: "confirmed", watcher: "finalized", isReobservation: true, result: true}, + + // Reobservation case that never really happen because only the finalized watcher does reobservations + {commitment: "finalized", watcher: "confirmed", isReobservation: true, result: false}, + {commitment: "confirmed", watcher: "confirmed", isReobservation: true, result: true}, + } + + for _, tc := range tests { + var label string + if tc.isReobservation { + label = "reobserved_" + } else { + label = "new_" + } + label += tc.commitment + "_message_on_" + tc.watcher + "_watcher" + t.Run(label, func(t *testing.T) { + commitment := rpc.CommitmentType(tc.commitment) + watcher := rpc.CommitmentType(tc.watcher) + s := &SolanaWatcher{commitment: watcher} + assert.Equal(t, tc.result, s.checkCommitment(commitment, tc.isReobservation)) + }) + } +}