fix(ws): pubsub parity (#407)

* fix(ws): pubsub parity

Cross-checked every response and request shape in rpc/ws against
agave/pubsub-client (anza-xyz/agave) and filled the gaps. Method-level
parity was already complete, but option sets, response fields, and the
BlockSubscribe/ParsedBlockSubscribe split had drifted in ways that
produced the "unable to decode client response" error on jsonParsed
block notifications (issue #291) and left several subscriptions without
the options Agave exposes.

BlockSubscribe unification (closes #291)
----------------------------------------
BlockSubscribe now accepts every UiTransactionEncoding the RPC
supports — base58, base64, base64+zstd, jsonParsed — through a single
method, matching Agave's shape. Go can't express
BlockResult.Value is a Go-style union: BlockResultValue carries both
Block (*rpc.GetBlockResult) and ParsedBlock (*rpc.GetParsedBlockResult).
Exactly one is populated per frame, chosen by the Encoding captured at
subscribe time and routed inside decodeBlockNotification(isParsed).

  base58 / base64 / base64+zstd  -> Block
  jsonParsed                     -> ParsedBlock

ParsedBlockSubscribe is marked Deprecated but continues to return
*ParsedBlockResult so nothing breaks for existing callers.

Request / config parity
-----------------------
- AccountSubscribeConfig: new struct mirroring RpcAccountInfoConfig
  (commitment, encoding, dataSlice, minContextSlot). The existing
  AccountSubscribe / AccountSubscribeWithOpts are preserved as thin
  wrappers; AccountSubscribeWithConfig exposes the full surface.
- ProgramSubscribeConfig: new struct mirroring RpcProgramAccountsConfig
  (commitment, encoding, dataSlice, filters, minContextSlot,
  withContext, sortResults).
- SignatureSubscribeConfig: new struct mirroring
  RpcSignatureSubscribeConfig, adds enableReceivedNotification.
- Each Config has an unexported (c *Config).params() helper that
  matches Agave's serde(skip_serializing_if="Option::is_none") behavior.

Response parity
---------------
- SignatureResult.Value now models the untagged
  ProcessedSignature | ReceivedSignature union via
  SignatureResultValue{IsReceived, Processed}. Custom UnmarshalJSON
  handles the object form ({"err":...}), the string form
  "receivedSignature", and null. The old Value.Err path is replaced
  by Value.Processed.Err.
- VoteResult gains VotePubkey and Signature to match RpcVote.
- SlotsUpdatesResult adds Err (Dead variant), documents which fields
  are populated per discriminator.
- All Context fields across the subscription result types moved from
  anonymous struct{Slot uint64} to a shared RPCResponseContext
  (slot + apiVersion), matching RpcResponseContext.

Consumer adaptation
-------------------
sendAndConfirmTransaction.WaitForConfirmation adapted to the new
SignatureResult shape and loops on IsReceived until a processed
notification arrives — previously retried only once, so a second
consecutive received-only frame would have passed a nil Processed
through and wrongly reported (confirmed=true, err=nil).

Examples
--------
- rpc/ws/examples/blockSubscribe: demonstrates the Block/ParsedBlock
  union via a switch on which field is populated.
- rpc/ws/examples/parsedBlockSubscribe: removed (redundant now).

Tests (new subscriptions_test.go + parsedBlockSubscribe_test.go)
----------------------------------------------------------------
Fixture JSON mirrors the wire format produced by Agave's
rpc_subscriptions.rs integration tests:
- accountNotification: base64 and jsonParsed payloads
- programNotification, logsNotification, voteNotification,
  slotNotification, rootNotification
- signatureNotification: processed (null err), processed (with err),
  and the "receivedSignature" string variant
- slotsUpdatesNotification: all seven SlotUpdate variants
- blockNotification: separate ParsedRoute / BinaryRoute tests that
  exercise decodeBlockNotification for both branches; the binary
  fixture includes a real [blob, "base64"] 2-array transaction +
  meta so drift between the two shapes would fail the test
- TestConfigParams_Wire verifies per-field emission and omit behavior
  on every *SubscribeConfig.

* chore: apply linting

* fix(message): use gojson

Author: sonicfromnewyoke

message.go

@@ -23,7 +23,7 @@ import (
 
 	bin "github.com/gagliardetto/binary"
 	"github.com/gagliardetto/treeout"
-	jsoniter "github.com/json-iterator/go"
+	gojson "github.com/goccy/go-json"
 
 	"github.com/gagliardetto/solana-go/text"
 )
@@ -242,7 +242,7 @@ func (mx *Message) UnmarshalJSON(data []byte) error {
 		Header              MessageHeader         `json:"header"`
 		RecentBlockhash     Hash                  `json:"recentBlockhash"`
 		Instructions        []CompiledInstruction `json:"instructions"`
-		AddressTableLookups *jsoniter.RawMessage  `json:"addressTableLookups"`
+		AddressTableLookups *gojson.RawMessage    `json:"addressTableLookups"`
 	}{}
 	if err := json.Unmarshal(data, &aux); err != nil {
 		return err

rpc/jsonrpc/jsonrpc.go

@@ -12,8 +12,8 @@ import (
 	"sync/atomic"
 
 	"github.com/davecgh/go-spew/spew"
-	stdjson "github.com/goccy/go-json"
 	gojson "github.com/goccy/go-json"
+	stdjson "github.com/goccy/go-json"
 	"github.com/google/uuid"
 )
 

rpc/sendAndConfirmTransaction/sendAndConfirmTransaction.go

@@ -123,15 +123,22 @@ func WaitForConfirmation(
 	timeoutCtx, cancel := context.WithTimeout(ctx, *timeout)
 	defer cancel()
 
-	got, err := sub.Recv(timeoutCtx)
-	if err != nil {
-		if timeoutCtx.Err() == context.DeadlineExceeded {
-			return false, ErrTimeout
+	// Drain any "receivedSignature" notifications (if the caller opted in
+	// to enableReceivedNotification) until we get the processed one.
+	for {
+		got, err := sub.Recv(timeoutCtx)
+		if err != nil {
+			if timeoutCtx.Err() == context.DeadlineExceeded {
+				return false, ErrTimeout
+			}
+			return false, err
 		}
-		return false, err
-	}
-	if got.Value.Err != nil {
-		return true, fmt.Errorf("confirmed transaction with execution error: %v", got.Value.Err)
+		if got.Value.IsReceived {
+			continue
+		}
+		if got.Value.Processed != nil && got.Value.Processed.Err != nil {
+			return true, fmt.Errorf("confirmed transaction with execution error: %v", got.Value.Processed.Err)
+		}
+		return true, nil
 	}
-	return true, nil
 }

rpc/util_test.go

@@ -11,19 +11,21 @@ import (
 // Layout references for the test data below:
 //
 // SPL Token (solana-program/token):
-//   Mint::LEN    = 82
-//   Account::LEN = 165
+//
+//	Mint::LEN    = 82
+//	Account::LEN = 165
 //
 // Token-2022 (solana-program/token-2022):
-//   Extended records place a 1-byte AccountType discriminator at offset
-//   Account::LEN (= 165). Mint base (82 bytes) is padded with 83 zeros so
-//   Mint and Account share the discriminator offset.
 //
-//   AccountType::Uninitialized = 0
-//   AccountType::Mint          = 1
-//   AccountType::Account       = 2
+//	Extended records place a 1-byte AccountType discriminator at offset
+//	Account::LEN (= 165). Mint base (82 bytes) is padded with 83 zeros so
+//	Mint and Account share the discriminator offset.
+//
+//	AccountType::Uninitialized = 0
+//	AccountType::Mint          = 1
+//	AccountType::Account       = 2
 //
-//   Extensions follow as TLV: [u16 LE type][u16 LE length][value...].
+//	Extensions follow as TLV: [u16 LE type][u16 LE length][value...].
 const (
 	testAccountTypeUninitialized uint8 = 0
 	testAccountTypeMint          uint8 = 1

rpc/ws/accountSubscribe.go

@@ -22,47 +22,78 @@ import (
 )
 
 type AccountResult struct {
-	Context struct {
-		Slot uint64 `json:"slot"`
-	} `json:"context"`
-	Value *rpc.Account `json:"value"`
+	Context RPCResponseContext `json:"context"`
+	Value   *rpc.Account       `json:"value"`
+}
+
+// AccountSubscribeConfig matches Agave's RpcAccountInfoConfig. It
+// supports every encoding the RPC does — base58, base64, base64+zstd,
+// and jsonParsed — via rpc.Account.Data, which is a union type
+// (rpc.DataBytesOrJSON) that decodes all of them transparently.
+type AccountSubscribeConfig struct {
+	Commitment     rpc.CommitmentType
+	Encoding       solana.EncodingType
+	DataSlice      *rpc.DataSlice
+	MinContextSlot *uint64
+}
+
+// params converts the config to the JSON-RPC params object the
+// accountSubscribe method expects. Missing options are omitted so the
+// wire format matches Agave's serde(skip_serializing_if) behavior.
+func (c *AccountSubscribeConfig) params() map[string]any {
+	conf := map[string]any{"encoding": solana.EncodingBase64}
+	if c == nil {
+		return conf
+	}
+	if c.Commitment != "" {
+		conf["commitment"] = c.Commitment
+	}
+	if c.Encoding != "" {
+		conf["encoding"] = c.Encoding
+	}
+	if c.DataSlice != nil {
+		conf["dataSlice"] = c.DataSlice
+	}
+	if c.MinContextSlot != nil {
+		conf["minContextSlot"] = *c.MinContextSlot
+	}
+	return conf
 }
 
 // AccountSubscribe subscribes to an account to receive notifications
-// when the lamports or data for a given account public key changes.
+// when its lamports or data change. Defaults to base64 encoding.
 func (cl *Client) AccountSubscribe(
 	account solana.PublicKey,
 	commitment rpc.CommitmentType,
 ) (*AccountSubscription, error) {
-	return cl.AccountSubscribeWithOpts(
-		account,
-		commitment,
-		"",
-	)
+	return cl.AccountSubscribeWithOpts(account, commitment, "")
 }
 
-// AccountSubscribe subscribes to an account to receive notifications
-// when the lamports or data for a given account public key changes.
+// AccountSubscribeWithOpts is the simple variant that accepts bare
+// commitment and encoding arguments.
+//
+// Deprecated: use AccountSubscribeWithConfig for the full option set
+// (dataSlice, minContextSlot) exposed by Agave's RpcAccountInfoConfig.
 func (cl *Client) AccountSubscribeWithOpts(
 	account solana.PublicKey,
 	commitment rpc.CommitmentType,
 	encoding solana.EncodingType,
 ) (*AccountSubscription, error) {
+	return cl.AccountSubscribeWithConfig(account, &AccountSubscribeConfig{
+		Commitment: commitment,
+		Encoding:   encoding,
+	})
+}
 
-	params := []any{account.String()}
-	conf := map[string]any{
-		"encoding": "base64",
-	}
-	if commitment != "" {
-		conf["commitment"] = commitment
-	}
-	if encoding != "" {
-		conf["encoding"] = encoding
-	}
-
+// AccountSubscribeWithConfig mirrors the full RpcAccountInfoConfig
+// option surface of the underlying accountSubscribe RPC.
+func (cl *Client) AccountSubscribeWithConfig(
+	account solana.PublicKey,
+	config *AccountSubscribeConfig,
+) (*AccountSubscription, error) {
 	genSub, err := cl.subscribe(
-		params,
-		conf,
+		[]any{account.String()},
+		config.params(),
 		"accountSubscribe",
 		"accountUnsubscribe",
 		func(msg []byte) (any, error) {
@@ -74,9 +105,7 @@ func (cl *Client) AccountSubscribeWithOpts(
 	if err != nil {
 		return nil, err
 	}
-	return &AccountSubscription{
-		sub: genSub,
-	}, nil
+	return &AccountSubscription{sub: genSub}, nil
 }
 
 type AccountSubscription struct {

rpc/ws/blockSubscribe.go

@@ -22,15 +22,34 @@ import (
 	"github.com/gagliardetto/solana-go/rpc"
 )
 
+// BlockResult is the unified notification payload for blockSubscribe.
+// Exactly one of Value.Block or Value.ParsedBlock is populated per
+// frame, based on the Encoding the caller passed at subscribe time.
 type BlockResult struct {
-	Context struct {
-		Slot uint64
-	} `json:"context"`
-	Value struct {
-		Slot  uint64              `json:"slot"`
-		Err   any                 `json:"err,omitempty"`
-		Block *rpc.GetBlockResult `json:"block,omitempty"`
-	} `json:"value"`
+	Context RPCResponseContext `json:"context"`
+	Value   BlockResultValue   `json:"value"`
+}
+
+// BlockResultValue mirrors Agave's RpcBlockUpdate with a Go-style union
+// over the two block shapes.
+//
+//	Encoding                              Field populated
+//	----------------------------------------------------
+//	base58 / base64 / base64+zstd         Block        (*rpc.GetBlockResult)
+//	jsonParsed                            ParsedBlock  (*rpc.GetParsedBlockResult)
+//
+// Err / Slot are always decoded regardless of which block field is set.
+type BlockResultValue struct {
+	Slot uint64 `json:"slot"`
+	Err  any    `json:"err,omitempty"`
+
+	// Block is populated for binary encodings.
+	Block *rpc.GetBlockResult `json:"block,omitempty"`
+
+	// ParsedBlock is populated when Encoding=jsonParsed was requested.
+	// BlockSubscribe sets this at decode time; it is not read by the
+	// default json.Unmarshal path.
+	ParsedBlock *rpc.GetParsedBlockResult `json:"-"`
 }
 
 type BlockSubscribeFilter interface {
@@ -76,7 +95,11 @@ type BlockSubscribeOpts struct {
 
 // NOTE: Unstable, disabled by default
 //
-// Subscribe to receive notification anytime a new block is Confirmed or Finalized.
+// BlockSubscribe subscribes to new blocks. Supports every encoding the
+// RPC does — base58, base64, base64+zstd, and jsonParsed. The
+// BlockResult.Value field is a union: the binary encodings populate
+// Block (*rpc.GetBlockResult); jsonParsed populates ParsedBlock
+// (*rpc.GetParsedBlockResult).
 //
 // **This subscription is unstable and only available if the validator was started
 // with the `--rpc-pubsub-enable-block-subscription` flag. The format of this
@@ -85,6 +108,8 @@ func (cl *Client) BlockSubscribe(
 	filter BlockSubscribeFilter,
 	opts *BlockSubscribeOpts,
 ) (*BlockSubscription, error) {
+	isParsed := opts != nil && opts.Encoding == solana.EncodingJSONParsed
+
 	var params []any
 	if filter != nil {
 		switch v := filter.(type) {
@@ -94,6 +119,7 @@ func (cl *Client) BlockSubscribe(
 			params = append(params, rpc.M{"mentionsAccountOrProgram": v.Pubkey})
 		}
 	}
+
 	if opts != nil {
 		obj := make(rpc.M)
 		if opts.Commitment != "" {
@@ -102,12 +128,10 @@ func (cl *Client) BlockSubscribe(
 		if opts.Encoding != "" {
 			if !solana.IsAnyOfEncodingType(
 				opts.Encoding,
-				// Valid encodings:
-				// solana.EncodingJSON, // TODO
-				solana.EncodingJSONParsed, // TODO
 				solana.EncodingBase58,
 				solana.EncodingBase64,
 				solana.EncodingBase64Zstd,
+				solana.EncodingJSONParsed,
 			) {
 				return nil, fmt.Errorf("provided encoding is not supported: %s", opts.Encoding)
 			}
@@ -126,16 +150,13 @@ func (cl *Client) BlockSubscribe(
 			params = append(params, obj)
 		}
 	}
+
 	genSub, err := cl.subscribe(
 		params,
 		nil,
 		"blockSubscribe",
 		"blockUnsubscribe",
-		func(msg []byte) (any, error) {
-			var res BlockResult
-			err := decodeResponseFromMessage(msg, &res)
-			return &res, err
-		},
+		decodeBlockNotification(isParsed),
 	)
 	if err != nil {
 		return nil, err
@@ -145,6 +166,42 @@ func (cl *Client) BlockSubscribe(
 	}, nil
 }
 
+// decodeBlockNotification returns a frame decoder that lands the block
+// either in BlockResultValue.Block (binary) or .ParsedBlock (jsonParsed).
+// Unexported but package-local so tests can exercise both branches
+// without a live server.
+func decodeBlockNotification(isParsed bool) func([]byte) (any, error) {
+	return func(msg []byte) (any, error) {
+		if isParsed {
+			var tmp struct {
+				Context RPCResponseContext `json:"context"`
+				Value   struct {
+					Slot  uint64                    `json:"slot"`
+					Err   any                       `json:"err,omitempty"`
+					Block *rpc.GetParsedBlockResult `json:"block,omitempty"`
+				} `json:"value"`
+			}
+			if err := decodeResponseFromMessage(msg, &tmp); err != nil {
+				return nil, err
+			}
+			return &BlockResult{
+				Context: tmp.Context,
+				Value: BlockResultValue{
+					Slot:        tmp.Value.Slot,
+					Err:         tmp.Value.Err,
+					ParsedBlock: tmp.Value.Block,
+				},
+			}, nil
+		}
+
+		var res BlockResult
+		if err := decodeResponseFromMessage(msg, &res); err != nil {
+			return nil, err
+		}
+		return &res, nil
+	}
+}
+
 type BlockSubscription struct {
 	sub *Subscription
 }