Add commitment level to VAAs

This allows requesting attestations for various commitment/confirmation levels. This is helpful for low-latency applications like Pyth.

Change-Id: Ib49ace163365106b227613d2f66b787b3e5f5461

bridge/cmd/guardiand/bridge.go

@@ -51,9 +51,8 @@ var (
 	bridgeKeyPath       *string
 	solanaBridgeAddress *string
 
-	ethRPC           *string
-	ethContract      *string
-	ethConfirmations *uint64
+	ethRPC      *string
+	ethContract *string
 
 	terraSupport  *bool
 	terraWS       *string
@@ -92,7 +91,6 @@ func init() {
 
 	ethRPC = BridgeCmd.Flags().String("ethRPC", "", "Ethereum RPC URL")
 	ethContract = BridgeCmd.Flags().String("ethContract", "", "Ethereum bridge contract address")
-	ethConfirmations = BridgeCmd.Flags().Uint64("ethConfirmations", 15, "Ethereum confirmation count requirement")
 
 	terraSupport = BridgeCmd.Flags().Bool("terra", false, "Turn on support for Terra")
 	terraWS = BridgeCmd.Flags().String("terraWS", "", "Path to terrad root for websocket connection")
@@ -396,7 +394,7 @@ func runBridge(cmd *cobra.Command, args []string) {
 		}
 
 		if err := supervisor.Run(ctx, "ethwatch",
-			ethereum.NewEthBridgeWatcher(*ethRPC, ethContractAddr, *ethConfirmations, lockC, setC).Run); err != nil {
+			ethereum.NewEthBridgeWatcher(*ethRPC, ethContractAddr, lockC, setC).Run); err != nil {
 			return err
 		}
 

bridge/pkg/common/chainlock.go

@@ -11,10 +11,11 @@ type MessagePublication struct {
 	TxHash    common.Hash // TODO: rename to identifier? on Solana, this isn't actually the tx hash
 	Timestamp time.Time
 
-	Nonce          uint32
-	Sequence       uint64
-	EmitterChain   vaa.ChainID
-	EmitterAddress vaa.Address
-	Payload        []byte
-	Persist        bool
+	Nonce            uint32
+	Sequence         uint64
+	ConsistencyLevel uint8
+	EmitterChain     vaa.ChainID
+	EmitterAddress   vaa.Address
+	Payload          []byte
+	Persist          bool
 }

bridge/pkg/ethereum/watcher.go

@@ -69,9 +69,8 @@ func init() {
 
 type (
 	EthBridgeWatcher struct {
-		url              string
-		bridge           eth_common.Address
-		minConfirmations uint64
+		url    string
+		bridge eth_common.Address
 
 		pendingLocks      map[eth_common.Hash]*pendingLock
 		pendingLocksGuard sync.Mutex
@@ -86,8 +85,8 @@ type (
 	}
 )
 
-func NewEthBridgeWatcher(url string, bridge eth_common.Address, minConfirmations uint64, lockEvents chan *common.MessagePublication, setEvents chan *common.GuardianSet) *EthBridgeWatcher {
-	return &EthBridgeWatcher{url: url, bridge: bridge, minConfirmations: minConfirmations, lockChan: lockEvents, setChan: setEvents, pendingLocks: map[eth_common.Hash]*pendingLock{}}
+func NewEthBridgeWatcher(url string, bridge eth_common.Address, lockEvents chan *common.MessagePublication, setEvents chan *common.GuardianSet) *EthBridgeWatcher {
+	return &EthBridgeWatcher{url: url, bridge: bridge, lockChan: lockEvents, setChan: setEvents, pendingLocks: map[eth_common.Hash]*pendingLock{}}
 }
 
 func (e *EthBridgeWatcher) Run(ctx context.Context) error {
@@ -180,14 +179,15 @@ func (e *EthBridgeWatcher) Run(ctx context.Context) error {
 				}
 
 				lock := &common.MessagePublication{
-					TxHash:         ev.Raw.TxHash,
-					Timestamp:      time.Unix(int64(b.Time()), 0),
-					Nonce:          ev.Nonce,
-					Sequence:       ev.Sequence,
-					EmitterChain:   vaa.ChainIDEthereum,
-					EmitterAddress: PadAddress(ev.Sender),
-					Payload:        ev.Payload,
-					Persist:        ev.PersistMessage,
+					TxHash:           ev.Raw.TxHash,
+					Timestamp:        time.Unix(int64(b.Time()), 0),
+					Nonce:            ev.Nonce,
+					Sequence:         ev.Sequence,
+					EmitterChain:     vaa.ChainIDEthereum,
+					EmitterAddress:   PadAddress(ev.Sender),
+					Payload:          ev.Payload,
+					Persist:          ev.PersistMessage,
+					ConsistencyLevel: ev.Commitment,
 				}
 
 				logger.Info("found new lockup transaction", zap.Stringer("tx", ev.Raw.TxHash),
@@ -259,7 +259,7 @@ func (e *EthBridgeWatcher) Run(ctx context.Context) error {
 				for hash, pLock := range e.pendingLocks {
 
 					// Transaction was dropped and never picked up again
-					if pLock.height+4*e.minConfirmations <= blockNumberU {
+					if pLock.height+4*uint64(pLock.lock.ConsistencyLevel) <= blockNumberU {
 						logger.Debug("lockup timed out", zap.Stringer("tx", pLock.lock.TxHash),
 							zap.Stringer("block", ev.Number))
 						delete(e.pendingLocks, hash)
@@ -267,7 +267,7 @@ func (e *EthBridgeWatcher) Run(ctx context.Context) error {
 					}
 
 					// Transaction is now ready
-					if pLock.height+e.minConfirmations <= ev.Number.Uint64() {
+					if pLock.height+uint64(pLock.lock.ConsistencyLevel) <= ev.Number.Uint64() {
 						logger.Debug("lockup confirmed", zap.Stringer("tx", pLock.lock.TxHash),
 							zap.Stringer("block", ev.Number))
 						delete(e.pendingLocks, hash)

bridge/pkg/processor/message.go

@@ -65,6 +65,7 @@ func (p *Processor) handleLockup(ctx context.Context, k *common.MessagePublicati
 		EmitterAddress:   k.EmitterAddress,
 		Payload:          k.Payload,
 		Sequence:         k.Sequence,
+		ConsistencyLevel: k.ConsistencyLevel,
 	}
 
 	// Generate digest of the unsigned VAA.

bridge/pkg/processor/observation.go

@@ -213,6 +213,7 @@ func (p *Processor) handleObservation(ctx context.Context, m *gossipv1.SignedObs
 			EmitterChain:     v.EmitterChain,
 			EmitterAddress:   v.EmitterAddress,
 			Payload:          v.Payload,
+			ConsistencyLevel: v.ConsistencyLevel,
 		}
 
 		// 2/3+ majority required for VAA to be valid - wait until we have quorum to submit VAA.

bridge/pkg/solana/client.go

@@ -49,7 +49,7 @@ var (
 		prometheus.HistogramOpts{
 			Name: "wormhole_solana_query_latency",
 			Help: "Latency histogram for Solana RPC calls",
-		}, []string{"operation"})
+		}, []string{"operation", "commitment"})
 )
 
 func init() {
@@ -90,7 +90,7 @@ func (s *SolanaWatcher) Run(ctx context.Context) error {
 					defer cancel()
 					start := time.Now()
 					slot, err := rpcClient.GetSlot(rCtx, "")
-					queryLatency.WithLabelValues("get_slot").Observe(time.Since(start).Seconds())
+					queryLatency.WithLabelValues("get_slot", "processed").Observe(time.Since(start).Seconds())
 					if err != nil {
 						solanaConnectionErrors.WithLabelValues("get_slot_error").Inc()
 						errC <- err
@@ -109,7 +109,8 @@ func (s *SolanaWatcher) Run(ctx context.Context) error {
 					defer cancel()
 					start = time.Now()
 
-					accounts, err := rpcClient.GetProgramAccounts(rCtx, s.bridge, &rpc.GetProgramAccountsOpts{
+					// Get finalized accounts
+					fAccounts, err := rpcClient.GetProgramAccounts(rCtx, s.bridge, &rpc.GetProgramAccountsOpts{
 						Commitment: rpc.CommitmentMax, // TODO: deprecated, use Finalized
 						Filters: []rpc.RPCFilter{
 							{
@@ -126,19 +127,65 @@ func (s *SolanaWatcher) Run(ctx context.Context) error {
 							},
 							{
 								Memcmp: &rpc.RPCFilterMemcmp{
-									Offset: 5,                         // Offset of VaaTime
+									Offset: 5,                 // Start of the ConsistencyLevel value
+									Bytes:  solana.Base58{32}, // Only grab messages that require max confirmations
+								},
+							},
+							{
+								Memcmp: &rpc.RPCFilterMemcmp{
+									Offset: 6,                         // Offset of VaaTime
 									Bytes:  solana.Base58{0, 0, 0, 0}, // This means this VAA hasn't been signed yet
 								},
 							},
 						},
 					})
-					queryLatency.WithLabelValues("get_program_accounts").Observe(time.Since(start).Seconds())
+					queryLatency.WithLabelValues("get_program_accounts", "max").Observe(time.Since(start).Seconds())
 					if err != nil {
 						solanaConnectionErrors.WithLabelValues("get_program_account_error").Inc()
 						errC <- err
 						return
 					}
 
+					// Get confirmed accounts
+					cAccounts, err := rpcClient.GetProgramAccounts(rCtx, s.bridge, &rpc.GetProgramAccountsOpts{
+						Commitment: rpc.CommitmentSingle, // TODO: deprecated, use Confirmed
+						Filters: []rpc.RPCFilter{
+							{
+								Memcmp: &rpc.RPCFilterMemcmp{
+									Offset: 0,                            // Start of the account
+									Bytes:  solana.Base58{'m', 's', 'g'}, // Prefix of the posted message accounts
+								},
+							},
+							{
+								Memcmp: &rpc.RPCFilterMemcmp{
+									Offset: 4,                   // Start of the Persist flag
+									Bytes:  solana.Base58{0x01}, // Only grab messages that need to be persisted
+								},
+							},
+							{
+								Memcmp: &rpc.RPCFilterMemcmp{
+									Offset: 5,                // Start of the ConsistencyLevel value
+									Bytes:  solana.Base58{1}, // Only grab messages that require the Confirmed level
+								},
+							},
+							{
+								Memcmp: &rpc.RPCFilterMemcmp{
+									Offset: 6,                         // Offset of VaaTime
+									Bytes:  solana.Base58{0, 0, 0, 0}, // This means this VAA hasn't been signed yet
+								},
+							},
+						},
+					})
+					queryLatency.WithLabelValues("get_program_accounts", "single").Observe(time.Since(start).Seconds())
+					if err != nil {
+						solanaConnectionErrors.WithLabelValues("get_program_account_error").Inc()
+						errC <- err
+						return
+					}
+
+					// Merge accounts
+					accounts := append(fAccounts, cAccounts...)
+
 					logger.Debug("fetched transfer proposals without VAA",
 						zap.Int("n", len(accounts)),
 						zap.Duration("took", time.Since(start)),
@@ -166,14 +213,15 @@ func (s *SolanaWatcher) Run(ctx context.Context) error {
 						copy(txHash[:], acc.Pubkey[:])
 
 						lock := &common.MessagePublication{
-							TxHash:         txHash,
-							Timestamp:      time.Unix(int64(proposal.SubmissionTime), 0),
-							Nonce:          proposal.Nonce,
-							Sequence:       proposal.Sequence,
-							EmitterChain:   vaa.ChainIDSolana,
-							EmitterAddress: proposal.EmitterAddress,
-							Payload:        proposal.Payload,
-							Persist:        proposal.Persist == 1,
+							TxHash:           txHash,
+							Timestamp:        time.Unix(int64(proposal.SubmissionTime), 0),
+							Nonce:            proposal.Nonce,
+							Sequence:         proposal.Sequence,
+							EmitterChain:     vaa.ChainIDSolana,
+							EmitterAddress:   proposal.EmitterAddress,
+							Payload:          proposal.Payload,
+							Persist:          proposal.Persist == 1,
+							ConsistencyLevel: proposal.ConsistencyLevel,
 						}
 
 						solanaLockupsConfirmed.Inc()
@@ -198,6 +246,7 @@ type (
 		VaaVersion uint8
 		// Borsh does not seem to support booleans, so 0=false / 1=true
 		Persist             uint8
+		ConsistencyLevel    uint8
 		VaaTime             uint32
 		VaaSignatureAccount vaa.Address
 		SubmissionTime      uint32

bridge/pkg/solana/submitter.go

@@ -132,6 +132,7 @@ func (e *SolanaVAASubmitter) Run(ctx context.Context) error {
 					Payload:          v.Payload,
 					GuardianSetIndex: v.GuardianSetIndex,
 					Signatures:       signatures,
+					ConsistencyLevel: uint32(v.ConsistencyLevel),
 				}, SkipPreflight: e.skipPreflight})
 				cancel()
 				if err != nil {

bridge/pkg/terra/watcher.go

@@ -211,14 +211,15 @@ func (e *BridgeWatcher) Run(ctx context.Context) error {
 				}
 
 				messagePublication := &common.MessagePublication{
-					TxHash:         txHashValue,
-					Timestamp:      time.Unix(blockTime.Int(), 0),
-					Nonce:          uint32(nonce.Uint()),
-					Sequence:       sequence.Uint(),
-					EmitterChain:   vaa.ChainIDTerra,
-					EmitterAddress: senderAddress,
-					Payload:        payloadValue,
-					Persist:        persist.Bool(),
+					TxHash:           txHashValue,
+					Timestamp:        time.Unix(blockTime.Int(), 0),
+					Nonce:            uint32(nonce.Uint()),
+					Sequence:         sequence.Uint(),
+					EmitterChain:     vaa.ChainIDTerra,
+					EmitterAddress:   senderAddress,
+					Payload:          payloadValue,
+					Persist:          persist.Bool(),
+					ConsistencyLevel: 0, // Instant finality
 				}
 				e.msgChan <- messagePublication
 				terraLockupsConfirmed.Inc()

bridge/pkg/vaa/structs.go

@@ -29,6 +29,8 @@ type (
 		Nonce uint32
 		// Sequence of the VAA
 		Sequence uint64
+		/// ConsistencyLevel of the VAA
+		ConsistencyLevel uint8
 		// EmitterChain the VAA was emitted on
 		EmitterChain ChainID
 		// EmitterAddress of the contract that emitted the Message
@@ -143,6 +145,15 @@ func Unmarshal(data []byte) (*VAA, error) {
 	if n, err := reader.Read(emitterAddress[:]); err != nil || n != 32 {
 		return nil, fmt.Errorf("failed to read emitter address [%d]: %w", n, err)
 	}
+	v.EmitterAddress = emitterAddress
+
+	if err := binary.Read(reader, binary.BigEndian, &v.Sequence); err != nil {
+		return nil, fmt.Errorf("failed to read sequence: %w", err)
+	}
+
+	if err := binary.Read(reader, binary.BigEndian, &v.ConsistencyLevel); err != nil {
+		return nil, fmt.Errorf("failed to read commitment: %w", err)
+	}
 
 	payload := make([]byte, 1000)
 	n, err := reader.Read(payload)
@@ -231,6 +242,7 @@ func (v *VAA) serializeBody() ([]byte, error) {
 	MustWrite(buf, binary.BigEndian, v.EmitterChain)
 	buf.Write(v.EmitterAddress[:])
 	MustWrite(buf, binary.BigEndian, v.Sequence)
+	MustWrite(buf, binary.BigEndian, v.ConsistencyLevel)
 	buf.Write(v.Payload)
 
 	return buf.Bytes(), nil

bridge/pkg/vaa/types_test.go

@@ -7,7 +7,6 @@ import (
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/crypto"
 	"github.com/stretchr/testify/require"
-	"math/big"
 	"testing"
 	"time"
 )
@@ -18,7 +17,7 @@ func TestSerializeDeserialize(t *testing.T) {
 		vaa  *VAA
 	}{
 		{
-			name: "BodyTransfer",
+			name: "NormalVAA",
 			vaa: &VAA{
 				Version:          1,
 				GuardianSetIndex: 9,
@@ -28,55 +27,13 @@ func TestSerializeDeserialize(t *testing.T) {
 						Signature: [65]byte{},
 					},
 				},
-				Timestamp: time.Unix(2837, 0),
-				Payload: &BodyTransfer{
-					Nonce:         38,
-					SourceChain:   2,
-					TargetChain:   1,
-					SourceAddress: Address{2, 1, 4},
-					TargetAddress: Address{2, 1, 3},
-					Asset: &AssetMeta{
-						Chain:   9,
-						Address: Address{9, 2, 4},
-					},
-					Amount: big.NewInt(29),
-				},
-			},
-		},
-		{
-			name: "GuardianSetUpdate",
-			vaa: &VAA{
-				Version:          1,
-				GuardianSetIndex: 9,
-				Signatures: []*Signature{
-					{
-						Index:     1,
-						Signature: [65]byte{},
-					},
-				},
-				Timestamp: time.Unix(2837, 0),
-				Payload: &BodyGuardianSetUpdate{
-					Keys:     []common.Address{{}, {}},
-					NewIndex: 2,
-				},
-			},
-		},
-		{
-			name: "ContractUpgrade",
-			vaa: &VAA{
-				Version:          1,
-				GuardianSetIndex: 9,
-				Signatures: []*Signature{
-					{
-						Index:     1,
-						Signature: [65]byte{},
-					},
-				},
-				Timestamp: time.Unix(2837, 0),
-				Payload: &BodyContractUpgrade{
-					ChainID:     ChainIDEthereum,
-					NewContract: Address{1, 3, 4, 5, 2, 3},
-				},
+				Timestamp:        time.Unix(2837, 0),
+				Nonce:            10,
+				Sequence:         3,
+				ConsistencyLevel: 5,
+				EmitterChain:     8,
+				EmitterAddress:   Address{1, 2, 3},
+				Payload:          []byte("abc"),
 			},
 		},
 	}
@@ -100,16 +57,12 @@ func TestVerifySignature(t *testing.T) {
 		Version:          8,
 		GuardianSetIndex: 9,
 		Timestamp:        time.Unix(2837, 0),
-		Payload: &BodyTransfer{
-			SourceChain:   2,
-			TargetChain:   1,
-			TargetAddress: Address{2, 1, 3},
-			Asset: &AssetMeta{
-				Chain:   9,
-				Address: Address{9, 2, 4},
-			},
-			Amount: big.NewInt(29),
-		},
+		Nonce:            5,
+		Sequence:         10,
+		ConsistencyLevel: 2,
+		EmitterChain:     2,
+		EmitterAddress:   Address{0, 1, 2, 3, 4},
+		Payload:          []byte("abcd"),
 	}
 
 	data, err := v.SigningMsg()

design/0001_generic_message_passing.md

@@ -133,6 +133,13 @@ VAA struct {
 	// Tracked per (EmitterChain, EmitterAddress) tuple.
 	Sequence uint64 // <-- NEW
 
+    // Level of consistency requested by the emitter.
+    //
+    // The semantic meaning of this field is specific to the target
+    // chain (like a commitment level on Solana, number of
+    // confirmations on Ethereum, or no meaning with instant finality). 
+    ConsistencyLevel uint8 // <-- NEW
+
 	// Payload of the message.
 	Payload []byte // <-- NEW
 }

design/0004_message_publishing.md

@@ -52,6 +52,10 @@ message submitted.
 The timestamp is derived by the guardian software using the finalized timestamp of the block the message was published
 in.
 
+When a message is posted, the emitter can specify for how many confirmations the guardians should wait before an
+attestation is produced. This allows latency sensitive applications to make sacrifices on safety while critical
+applications can sacrifice latency over safety. Chains with instant finality can omit the argument.
+
 **Persistence:**
 
 In case an emitter chooses to persist a message, the guardian software will publish it to the Solana blockchain where
@@ -89,7 +93,7 @@ bridge tokens back to the chain where the governance and staking contracts are l
 
 Proposed bridge interface:
 
-`postMessage(bytes payload, bool persist)` - Publish a message to be attested by Wormhole.
+`postMessage(bytes payload, bool persist, u8 confirmations)` - Publish a message to be attested by Wormhole.
 
 `postVAA(VAA signed_vaa)` - Persist a VAA on chain (Solana only)
 

devnet/bridge.yaml

@@ -76,8 +76,6 @@ spec:
             - /tmp/terra.key
             - --agentRPC
             - /run/bridge/agent.sock
-            - --ethConfirmations
-            - '2'
             - --solanaBridgeAddress
             - Bridge1p5gheXUvJ6jGWGeCsgPKgnE3YgdGKRVCMY9o
             - --solanaWS

ethereum/contracts/Implementation.sol

@@ -9,13 +9,14 @@ import "./Governance.sol";
 import "@openzeppelin/contracts/proxy/ERC1967/ERC1967Upgrade.sol";
 
 contract Implementation is Governance {
-    event LogMessagePublished(address indexed sender, uint64 sequence, uint32 nonce, bytes payload, bool persistMessage);
+    event LogMessagePublished(address indexed sender, uint64 sequence, uint32 nonce, bytes payload, bool persistMessage, uint8 consistency_level);
 
     // Publish a message to be attested by the Wormhole network
     function publishMessage(
         uint32 nonce,
         bytes memory payload,
-        bool persistMessage
+        bool persistMessage,
+        uint8 consistency_level
     ) public payable {
         // check fee
         if( persistMessage ) {
@@ -25,7 +26,7 @@ contract Implementation is Governance {
         }
 
         // emit log
-        emit LogMessagePublished(msg.sender, useSequence(msg.sender), nonce, payload, persistMessage);
+        emit LogMessagePublished(msg.sender, useSequence(msg.sender), nonce, payload, persistMessage, consistency_level);
     }
 
     function useSequence(address emitter) internal returns (uint64 sequence) {

ethereum/contracts/Messages.sol

@@ -101,6 +101,9 @@ contract Messages is Getters {
         vm.sequence = encodedVM.toUint64(index);
         index += 8;
 
+        vm.consistencyLevel = encodedVM.toUint8(index);
+        index += 1;
+
         vm.payload = encodedVM.slice(index, encodedVM.length - index);
     }
 }

ethereum/contracts/Structs.sol

@@ -29,6 +29,7 @@ interface Structs {
 		uint16 emitterChainId;
 		bytes32 emitterAddress;
 		uint64 sequence;
+		uint8 consistencyLevel;
 		bytes payload;
 
 		uint32 guardianSetIndex;

proto/agent/v1/service.proto

@@ -27,9 +27,10 @@ message VAA {
   uint32 EmitterChain = 4;
   bytes EmitterAddress = 5;
   uint64 Sequence = 6;
-  bytes Payload = 7;
-  uint32 GuardianSetIndex = 8;
-  repeated Signature Signatures = 9;
+  uint32 ConsistencyLevel = 7;
+  bytes Payload = 8;
+  uint32 GuardianSetIndex = 9;
+  repeated Signature Signatures = 10;
 }
 
 message Signature{

solana/bridge/agent/src/main.rs

@@ -125,6 +125,7 @@ impl Agent for AgentImpl {
                 emitter_chain: vaa.emitter_chain as u16,
                 emitter_address: emitter_address,
                 sequence: vaa.sequence,
+                consistency_level: vaa.consistency_level as u8,
                 payload: vaa.payload.clone(),
             };
 

solana/bridge/client/src/main.rs

@@ -114,6 +114,7 @@ fn command_post_message(
     nonce: u32,
     payload: Vec<u8>,
     persist: bool,
+    commitment: bridge::types::ConsistencyLevel,
 ) -> CommmandResult {
     println!("Posting a message to the wormhole");
 
@@ -138,13 +139,16 @@ fn command_post_message(
         &FeeCollector::key(None, bridge),
         fee,
     );
+
+    let emitter = Keypair::new();
     let (_, ix) = bridge::instructions::post_message(
         *bridge,
         config.owner.pubkey(),
-        config.fee_payer.pubkey(),
+        emitter.pubkey(),
         nonce,
         payload,
         persist,
+        commitment,
     )
     .unwrap();
     let mut transaction =
@@ -152,7 +156,10 @@ fn command_post_message(
 
     let (recent_blockhash, fee_calculator) = config.rpc_client.get_recent_blockhash()?;
     check_fee_payer_balance(config, fee_calculator.calculate_fee(&transaction.message()))?;
-    transaction.sign(&[&config.fee_payer, &config.owner], recent_blockhash);
+    transaction.sign(
+        &[&config.fee_payer, &config.owner, &emitter],
+        recent_blockhash,
+    );
     Ok(Some(transaction))
 }
 
@@ -279,12 +286,20 @@ fn main() {
                         .required(true)
                         .help("Nonce of the message"),
                 )
+                .arg(
+                    Arg::with_name("consistency_level")
+                        .value_name("CONSISTENCY_LEVEL")
+                        .takes_value(true)
+                        .index(3)
+                        .required(true)
+                        .help("Consistency (Commitment) level at which the VAA should be produced <FINALIZED|CONFIRMED>"),
+                )
                 .arg(
                     Arg::with_name("data")
                         .validator(is_hex)
                         .value_name("DATA")
                         .takes_value(true)
-                        .index(3)
+                        .index(4)
                         .required(true)
                         .help("Payload of the message"),
                 )
@@ -352,8 +367,23 @@ fn main() {
             let data = hex::decode(data_str).unwrap();
             let nonce: u32 = value_of(arg_matches, "nonce").unwrap();
             let persist = arg_matches.is_present("persist");
+            let consistency_level: String = value_of(arg_matches, "consistency_level").unwrap();
 
-            command_post_message(&config, &bridge, nonce, data, persist)
+            command_post_message(
+                &config,
+                &bridge,
+                nonce,
+                data,
+                persist,
+                match consistency_level.to_lowercase().as_str() {
+                    "finalized" => bridge::types::ConsistencyLevel::Finalized,
+                    "confirmed" => bridge::types::ConsistencyLevel::Confirmed,
+                    _ => {
+                        eprintln!("Invalid commitment level");
+                        exit(1);
+                    }
+                },
+            )
         }
 
         _ => unreachable!(),

solana/bridge/program/src/api/initialize.rs

@@ -17,7 +17,7 @@ use solitaire::{
 
 type Payer<'a> = Signer<Info<'a>>;
 
-#[derive(FromAccounts, ToInstruction)]
+#[derive(FromAccounts)]
 pub struct Initialize<'b> {
     /// Bridge config.
     pub bridge: Mut<Bridge<'b, { AccountState::Uninitialized }>>,

solana/bridge/program/src/api/post_message.rs

@@ -11,6 +11,7 @@ use crate::{
         InsufficientFees,
         MathOverflow,
     },
+    types::ConsistencyLevel,
     CHAIN_ID_SOLANA,
 };
 use solana_program::{
@@ -64,7 +65,7 @@ impl<'b> InstructionContext<'b> for PostMessage<'b> {
     }
 }
 
-#[derive(BorshDeserialize, BorshSerialize, Default)]
+#[derive(BorshDeserialize, BorshSerialize)]
 pub struct PostMessageData {
     /// Unique nonce for this message
     pub nonce: u32,
@@ -74,6 +75,9 @@ pub struct PostMessageData {
 
     /// Should the VAA for this message be persisted on-chain
     pub persist: bool,
+
+    /// Commitment Level required for an attestation to be produced
+    pub consistency_level: ConsistencyLevel,
 }
 
 pub fn post_message(
@@ -138,6 +142,10 @@ pub fn post_message(
     accs.message.payload = data.payload;
     accs.message.sequence = accs.sequence.sequence;
     accs.message.persist = data.persist;
+    accs.message.consistency_level = match data.consistency_level {
+        ConsistencyLevel::Confirmed => 1,
+        ConsistencyLevel::Finalized => 32,
+    };
 
     // Create message account
     accs.message

solana/bridge/program/src/api/post_vaa.rs

@@ -111,6 +111,7 @@ pub struct PostVAAData {
     pub emitter_chain: u16,
     pub emitter_address: ForeignAddress,
     pub sequence: u64,
+    pub consistency_level: u8,
     pub payload: Vec<u8>,
 }
 
@@ -167,6 +168,7 @@ pub fn post_vaa(ctx: &ExecutionContext, accs: &mut PostVAA, vaa: PostVAAData) ->
         accs.message.emitter_address = vaa.emitter_address;
         accs.message.sequence = vaa.sequence;
         accs.message.payload = vaa.payload;
+        accs.message.consistency_level = vaa.consistency_level;
         accs.message
             .create(&msg_derivation, ctx, accs.payer.key, Exempt)?;
     }
@@ -218,6 +220,7 @@ fn check_integrity<'r>(
         v.write_u16::<BigEndian>(vaa.emitter_chain)?;
         v.write(&vaa.emitter_address)?;
         v.write_u64::<BigEndian>(vaa.sequence)?;
+        v.write_u8(vaa.consistency_level)?;
         v.write(&vaa.payload)?;
         v.into_inner()
     };

solana/bridge/program/src/instructions.rs

@@ -28,6 +28,7 @@ use crate::{
         SignatureSet,
         SignatureSetDerivationData,
     },
+    types::ConsistencyLevel,
     InitializeData,
     PayloadMessage,
     PostMessageData,
@@ -83,6 +84,7 @@ pub fn post_message(
     nonce: u32,
     payload: Vec<u8>,
     persist: bool,
+    commitment: ConsistencyLevel,
 ) -> solitaire::Result<(Pubkey, Instruction)> {
     let bridge = Bridge::<'_, { AccountState::Uninitialized }>::key(None, &program_id);
     let fee_collector = FeeCollector::<'_>::key(None, &program_id);
@@ -124,6 +126,7 @@ pub fn post_message(
                 nonce,
                 payload: payload.clone(),
                 persist,
+                consistency_level: commitment,
             })
             .try_to_vec()?,
         },
@@ -389,6 +392,7 @@ pub fn serialize_vaa(vaa: &PostVAAData) -> Vec<u8> {
     v.write_u16::<BigEndian>(vaa.emitter_chain).unwrap();
     v.write(&vaa.emitter_address).unwrap();
     v.write_u64::<BigEndian>(vaa.sequence).unwrap();
+    v.write_u8(vaa.consistency_level).unwrap();
     v.write(&vaa.payload).unwrap();
     v.into_inner()
 }

solana/bridge/program/src/types.rs

@@ -167,6 +167,9 @@ pub struct PostedMessageData {
     /// Whether the VAA for this message should be persisted
     pub persist: bool,
 
+    /// Level of consistency requested by the emitter
+    pub consistency_level: u8,
+
     /// Time the vaa was submitted
     pub vaa_time: u32,
 
@@ -387,3 +390,10 @@ impl DeserializeGovernancePayload for GovernancePayloadTransferFees {
     const MODULE: &'static str = "CORE";
     const ACTION: u8 = 4;
 }
+
+#[repr(u8)]
+#[derive(BorshSerialize, BorshDeserialize, Clone)]
+pub enum ConsistencyLevel {
+    Confirmed,
+    Finalized,
+}

solana/bridge/program/tests/integration.rs

@@ -1,12 +1,12 @@
 #![allow(warnings)]
 
-use rand::Rng;
 use borsh::BorshSerialize;
 use byteorder::{
     BigEndian,
     WriteBytesExt,
 };
 use hex_literal::hex;
+use rand::Rng;
 use secp256k1::{
     Message as Secp256k1Message,
     PublicKey,
@@ -38,6 +38,10 @@ use solana_sdk::{
     },
     transaction::Transaction,
 };
+use solitaire::{
+    processors::seeded::Seeded,
+    AccountState,
+};
 use std::{
     convert::TryInto,
     io::{
@@ -49,10 +53,6 @@ use std::{
         SystemTime,
     },
 };
-use solitaire::{
-    processors::seeded::Seeded,
-    AccountState,
-};
 
 use bridge::{
     accounts::{
@@ -125,7 +125,7 @@ fn run_integration_tests() {
     let mut context = Context {
         public: public_keys,
         secret: secret_keys,
-        seq: Sequencer { 
+        seq: Sequencer {
             sequences: std::collections::HashMap::new(),
         },
     };
@@ -158,7 +158,8 @@ fn test_initialize(context: &mut Context) {
     let now = std::time::SystemTime::now()
         .duration_since(std::time::UNIX_EPOCH)
         .unwrap()
-        .as_secs() - 10;
+        .as_secs()
+        - 10;
 
     common::initialize(client, program, payer, &*context.public.clone(), 500, 5000);
     common::sync(client, payer);
@@ -213,7 +214,8 @@ fn test_bridge_messages(context: &mut Context) {
 
         // Emulate Guardian behaviour, verifying the data and publishing signatures/VAA.
         let (vaa, body, body_hash) = common::generate_vaa(&emitter, message.clone(), nonce, 0, 1);
-        common::verify_signatures(client, program, payer, body, body_hash, &context.secret, 0).unwrap();
+        common::verify_signatures(client, program, payer, body, body_hash, &context.secret, 0)
+            .unwrap();
         common::post_vaa(client, program, payer, vaa).unwrap();
         common::sync(client, payer);
 
@@ -246,7 +248,8 @@ fn test_bridge_messages(context: &mut Context) {
 
         for (signature, secret_key) in signatures.signatures.iter().zip(context.secret.iter()) {
             // Sign message locally.
-            let (local_sig, recover_id) = secp256k1::sign(&Secp256k1Message::parse(&body_hash), &secret_key);
+            let (local_sig, recover_id) =
+                secp256k1::sign(&Secp256k1Message::parse(&body_hash), &secret_key);
 
             // Combine recoverify with signature to match 65 byte layout.
             let mut signature_bytes = [0u8; 65];
@@ -311,7 +314,8 @@ fn test_bridge_messages(context: &mut Context) {
 
     for (signature, secret_key) in signatures.signatures.iter().zip(context.secret.iter()) {
         // Sign message locally.
-        let (local_sig, recover_id) = secp256k1::sign(&Secp256k1Message::parse(&body_hash), &secret_key);
+        let (local_sig, recover_id) =
+            secp256k1::sign(&Secp256k1Message::parse(&body_hash), &secret_key);
 
         // Combine recoverify with signature to match 65 byte layout.
         let mut signature_bytes = [0u8; 65];
@@ -342,7 +346,8 @@ fn test_persistent_bridge_messages(context: &mut Context) {
         message.clone(),
         4999,
         true,
-    ).is_err());
+    )
+    .is_err());
 
     // Check current balance to verify the right fee is going to be taken.
     let fee_collector = FeeCollector::key(None, &program);
@@ -400,7 +405,8 @@ fn test_persistent_bridge_messages(context: &mut Context) {
 
     for (signature, secret_key) in signatures.signatures.iter().zip(context.secret.iter()) {
         // Sign message locally.
-        let (local_sig, recover_id) = secp256k1::sign(&Secp256k1Message::parse(&body_hash), &secret_key);
+        let (local_sig, recover_id) =
+            secp256k1::sign(&Secp256k1Message::parse(&body_hash), &secret_key);
 
         // Combine recoverify with signature to match 65 byte layout.
         let mut signature_bytes = [0u8; 65];
@@ -454,7 +460,8 @@ fn test_invalid_emitter(context: &mut Context) {
             instruction,
         ],
         solana_sdk::commitment_config::CommitmentConfig::processed(),
-    ).is_err());
+    )
+    .is_err());
 }
 
 fn test_duplicate_messages_fail(context: &mut Context) {
@@ -489,7 +496,8 @@ fn test_duplicate_messages_fail(context: &mut Context) {
         message.clone(),
         10_000,
         false,
-    ).is_err());
+    )
+    .is_err());
 }
 
 fn test_guardian_set_change(context: &mut Context) {
@@ -500,7 +508,8 @@ fn test_guardian_set_change(context: &mut Context) {
     let now = std::time::SystemTime::now()
         .duration_since(std::time::UNIX_EPOCH)
         .unwrap()
-        .as_secs() - 10;
+        .as_secs()
+        - 10;
 
     // Upgrade the guardian set with a new set of guardians.
     let (new_public_keys, new_secret_keys) = common::generate_keys(1);
@@ -618,7 +627,8 @@ fn test_guardian_set_change(context: &mut Context) {
 
     for (signature, secret_key) in signatures.signatures.iter().zip(context.secret.iter()) {
         // Sign message locally.
-        let (local_sig, recover_id) = secp256k1::sign(&Secp256k1Message::parse(&body_hash), &secret_key);
+        let (local_sig, recover_id) =
+            secp256k1::sign(&Secp256k1Message::parse(&body_hash), &secret_key);
 
         // Combine recoverify with signature to match 65 byte layout.
         let mut signature_bytes = [0u8; 65];
@@ -801,7 +811,8 @@ fn test_set_fees(context: &mut Context) {
 
     for (signature, secret_key) in signatures.signatures.iter().zip(context.secret.iter()) {
         // Sign message locally.
-        let (local_sig, recover_id) = secp256k1::sign(&Secp256k1Message::parse(&body_hash), &secret_key);
+        let (local_sig, recover_id) =
+            secp256k1::sign(&Secp256k1Message::parse(&body_hash), &secret_key);
 
         // Combine recoverify with signature to match 65 byte layout.
         let mut signature_bytes = [0u8; 65];
@@ -851,7 +862,8 @@ fn test_set_fees_fails(context: &mut Context) {
         message_key,
         emitter.pubkey(),
         sequence,
-    ).is_err());
+    )
+    .is_err());
     common::sync(client, payer);
 }
 
@@ -961,7 +973,8 @@ fn test_free_fees(context: &mut Context) {
 
     for (signature, secret_key) in signatures.signatures.iter().zip(context.secret.iter()) {
         // Sign message locally.
-        let (local_sig, recover_id) = secp256k1::sign(&Secp256k1Message::parse(&body_hash), &secret_key);
+        let (local_sig, recover_id) =
+            secp256k1::sign(&Secp256k1Message::parse(&body_hash), &secret_key);
 
         // Combine recoverify with signature to match 65 byte layout.
         let mut signature_bytes = [0u8; 65];
@@ -1113,7 +1126,8 @@ fn test_transfer_too_much(context: &mut Context) {
         emitter.pubkey(),
         payer.pubkey(),
         sequence,
-    ).is_err());
+    )
+    .is_err());
 }
 
 fn test_foreign_bridge_messages(context: &mut Context) {
@@ -1170,7 +1184,8 @@ fn test_foreign_bridge_messages(context: &mut Context) {
 
     for (signature, secret_key) in signatures.signatures.iter().zip(context.secret.iter()) {
         // Sign message locally.
-        let (local_sig, recover_id) = secp256k1::sign(&Secp256k1Message::parse(&body_hash), &secret_key);
+        let (local_sig, recover_id) =
+            secp256k1::sign(&Secp256k1Message::parse(&body_hash), &secret_key);
 
         // Combine recoverify with signature to match 65 byte layout.
         let mut signature_bytes = [0u8; 65];

solana/solitaire/program/src/processors/peel.rs

@@ -63,8 +63,7 @@ impl<'a, 'b: 'a, 'c, T: Peel<'a, 'b, 'c>, const Seed: &'static str> Peel<'a, 'b,
 }
 
 /// Peel a Mutable key.
-impl<'a, 'b: 'a, 'c, T: Peel<'a, 'b, 'c>> Peel<'a, 'b, 'c> for Mut<T>
-{
+impl<'a, 'b: 'a, 'c, T: Peel<'a, 'b, 'c>> Peel<'a, 'b, 'c> for Mut<T> {
     fn peel<I>(mut ctx: &'c mut Context<'a, 'b, 'c, I>) -> Result<Self> {
         ctx.immutable = false;
         match ctx.info().is_writable {

terra/contracts/wormhole/src/state.rs

@@ -49,6 +49,7 @@ pub struct ParsedVAA {
     pub emitter_chain: u16,
     pub emitter_address: Vec<u8>,
     pub sequence: u64,
+    pub consistency_level: u8,
     pub payload: Vec<u8>,
 
     pub hash: Vec<u8>,
@@ -72,6 +73,7 @@ impl ParsedVAA {
     12  uint16      emitter_chain
     14  [32]uint8   emitter_address
     46  uint64      sequence
+    46  uint8       consistency_level
     54  []uint8     payload
     */
 
@@ -85,7 +87,8 @@ impl ParsedVAA {
     pub const VAA_EMITTER_CHAIN_POS: usize = 12;
     pub const VAA_EMITTER_ADDRESS_POS: usize = 14;
     pub const VAA_SEQUENCE_POS: usize = 46;
-    pub const VAA_PAYLOAD_POS: usize = 54;
+    pub const VAA_CONSISTENCY_LEVEL_POS: usize = 54;
+    pub const VAA_PAYLOAD_POS: usize = 55;
 
     // Signature data offsets in the signature block
     pub const SIG_DATA_POS: usize = 1;
@@ -123,6 +126,7 @@ impl ParsedVAA {
             .get_bytes32(body_offset + Self::VAA_EMITTER_ADDRESS_POS)
             .to_vec();
         let sequence = data.get_u64(body_offset + Self::VAA_SEQUENCE_POS);
+        let consistency_level = data.get_u8(body_offset + Self::VAA_CONSISTENCY_LEVEL_POS);
         let payload = data[body_offset + Self::VAA_PAYLOAD_POS..].to_vec();
 
         Ok(ParsedVAA {
@@ -134,6 +138,7 @@ impl ParsedVAA {
             emitter_chain,
             emitter_address,
             sequence,
+            consistency_level,
             payload,
             hash,
         })