feat: destroy stream in CRE flow

MicBun · MicBun · commit f0a7d9f695ac · 2025-12-24T18:39:15.000+07:00
diff --git a/core/tnclient/client.go b/core/tnclient/client.go
@@ -143,10 +143,21 @@ func (c *Client) DeployStream(ctx context.Context, streamId util.StreamId, strea
 }
 
 func (c *Client) DestroyStream(ctx context.Context, streamId util.StreamId) (types.Hash, error) {
-	return tn_api.DestroyStream(ctx, tn_api.DestroyStreamInput{
-		StreamId:   streamId,
-		KwilClient: c.GetKwilClient(),
-	})
+	// For HTTP transport, use the existing implementation (backwards compatible)
+	// For custom transports (CRE, etc.), use transport.Execute directly
+	if httpTransport, ok := c.transport.(*HTTPTransport); ok {
+		return tn_api.DestroyStream(ctx, tn_api.DestroyStreamInput{
+			StreamId:   streamId,
+			KwilClient: httpTransport.gatewayClient,
+		})
+	}
+	// Use transport.Execute directly for custom transports
+	// Derive address from signer for delete_stream call
+	addr, _ := auth.EthSecp256k1Authenticator{}.Identifier(c.signer.CompactID())
+	return c.transport.Execute(ctx, "", "delete_stream", [][]any{{
+		addr,
+		streamId.String(),
+	}})
 }
 
 func (c *Client) LoadActions() (clientType.IAction, error) {
diff --git a/core/tnclient/transport_cre.go b/core/tnclient/transport_cre.go
@@ -7,6 +7,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"math/big"
 	"net/url"
 	"regexp"
 	"strconv"
@@ -31,6 +32,14 @@ import (
 // workflows where standard net/http is not available. It uses CRE's consensus-aware
 // HTTP client with Promise-based async operations.
 //
+// KNOWN LIMITATION - CRE WASM Simulation Mode:
+// The CRE simulator has a bug where transaction signatures get corrupted due to
+// Go WASM pointer handling limitations (golang/go#59156, golang/go#66984).
+// Symptoms: "invalid signature: expected address X, received address Y"
+// This affects local testing with `cre workflow simulate` but does NOT affect
+// production deployments to Chainlink DON nodes. For local testing, use the
+// HTTP transport version (see examples/truf-cre-demo/test-http-version).
+//
 // Example usage in CRE workflow:
 //
 //	func onCronTrigger(config *Config, runtime cre.Runtime, trigger *cron.Payload) (*Result, error) {
@@ -63,6 +72,9 @@ type CRETransport struct {
 	reqID              atomic.Uint64
 	authCookie         string // Cookie value for gateway authentication
 	authCookieMu       sync.RWMutex
+	currentNonce       int64 // Track nonce for sequential transactions
+	nonceMu            sync.Mutex
+	nonceFetched       bool
 }
 
 // Verify CRETransport implements Transport interface at compile time
@@ -192,6 +204,19 @@ func (t *CRETransport) doJSONRPC(ctx context.Context, method string, params any,
 
 	// Check for JSON-RPC errors
 	if rpcResp.Error != nil {
+		// For broadcast errors (-201), decode the BroadcastError details
+		if rpcResp.Error.Code == -201 && len(rpcResp.Error.Data) > 0 {
+			var broadcastErr struct {
+				Code    uint32 `json:"code"`
+				Hash    string `json:"hash"`
+				Message string `json:"message"`
+			}
+			if err := json.Unmarshal(rpcResp.Error.Data, &broadcastErr); err == nil {
+				return fmt.Errorf("JSON-RPC error: %s (code: %d) [Broadcast: code=%d, hash=%s, msg=%s]",
+					rpcResp.Error.Message, rpcResp.Error.Code,
+					broadcastErr.Code, broadcastErr.Hash, broadcastErr.Message)
+			}
+		}
 		return fmt.Errorf("JSON-RPC error: %s (code: %d)", rpcResp.Error.Message, rpcResp.Error.Code)
 	}
 
@@ -259,12 +284,40 @@ func (t *CRETransport) Call(ctx context.Context, namespace string, action string
 //
 // This method builds a signed transaction and broadcasts it to the TRUF.NETWORK.
 // The transaction is signed using the configured signer and executed within CRE's
-// consensus mechanism.
+// consensus mechanism. Automatically retries on nonce errors.
 func (t *CRETransport) Execute(ctx context.Context, namespace string, action string, inputs [][]any, opts ...clientType.TxOpt) (types.Hash, error) {
 	if t.signer == nil {
 		return types.Hash{}, fmt.Errorf("signer required for Execute operations")
 	}
 
+	// Retry loop for nonce errors
+	const maxRetries = 3
+	for attempt := 0; attempt < maxRetries; attempt++ {
+		fmt.Printf("[DEBUG] Execute attempt %d/%d for action=%s\n", attempt+1, maxRetries, action)
+		txHash, err := t.executeOnce(ctx, namespace, action, inputs, opts...)
+		if err != nil {
+			fmt.Printf("[DEBUG] Execute error on attempt %d: %v\n", attempt+1, err)
+			// Check if it's a nonce error
+			if strings.Contains(err.Error(), "invalid nonce") && attempt < maxRetries-1 {
+				fmt.Printf("[DEBUG] Nonce error detected, resetting nonceFetched and retrying\n")
+				// Reset nonce tracking to refetch on next attempt
+				t.nonceMu.Lock()
+				t.nonceFetched = false
+				t.nonceMu.Unlock()
+				continue // Retry
+			}
+			return types.Hash{}, err
+		}
+		return txHash, nil
+	}
+
+	return types.Hash{}, fmt.Errorf("max retries exceeded")
+}
+
+// executeOnce performs a single execute attempt (internal helper)
+func (t *CRETransport) executeOnce(ctx context.Context, namespace string, action string, inputs [][]any, opts ...clientType.TxOpt) (types.Hash, error) {
+	fmt.Printf("[DEBUG] executeOnce called: action=%s\n", action)
+
 	// Convert inputs to EncodedValue arrays
 	var encodedInputs [][]*types.EncodedValue
 	for _, inputRow := range inputs {
@@ -298,6 +351,54 @@ func (t *CRETransport) Execute(ctx context.Context, namespace string, action str
 		opt(txOpts)
 	}
 
+	// Auto-manage nonce if not explicitly provided
+	if txOpts.Nonce == 0 {
+		t.nonceMu.Lock()
+
+		// Fetch nonce from gateway on first transaction only
+		if !t.nonceFetched {
+			// Create AccountID from signer
+			acctID := &types.AccountID{
+				Identifier: t.signer.CompactID(),
+				KeyType:    t.signer.PubKey().Type(),
+			}
+
+			// Fetch account info via user.account RPC call
+			params := map[string]any{
+				"id": acctID,
+			}
+
+			var accountResp struct {
+				ID      *types.AccountID `json:"id"`
+				Balance string           `json:"balance"`
+				Nonce   int64            `json:"nonce"`
+			}
+
+			err := t.callJSONRPC(ctx, "user.account", params, &accountResp)
+			if err != nil {
+				// If account doesn't exist yet, start with nonce 0
+				if !strings.Contains(err.Error(), "not found") && !strings.Contains(err.Error(), "does not exist") {
+					t.nonceMu.Unlock()
+					return types.Hash{}, fmt.Errorf("failed to fetch account nonce: %w", err)
+				}
+				t.currentNonce = 0
+				fmt.Printf("[DEBUG] Account not found, starting with nonce=0\n")
+			} else {
+				// Account nonce is the LAST used nonce, so NEXT nonce is nonce+1
+				t.currentNonce = accountResp.Nonce + 1
+				fmt.Printf("[DEBUG] Fetched account nonce=%d, using next nonce=%d\n", accountResp.Nonce, t.currentNonce)
+			}
+			t.nonceFetched = true
+		}
+
+		// Use current nonce and increment
+		txOpts.Nonce = t.currentNonce
+		fmt.Printf("[DEBUG] Using nonce=%d for transaction\n", t.currentNonce)
+		t.currentNonce++
+
+		t.nonceMu.Unlock()
+	}
+
 	// Ensure chain ID is fetched before building transaction
 	// This prevents transactions with empty chain IDs
 	// Check if already initialized (read lock)
@@ -322,33 +423,108 @@ func (t *CRETransport) Execute(ctx context.Context, namespace string, action str
 		t.chainIDMu.Unlock()
 	}
 
+	// Ensure Fee is not nil to prevent signature verification mismatch
+	// When Fee is nil, SerializeMsg produces "Fee: <nil>" but after JSON
+	// marshaling/unmarshaling it becomes "Fee: 0", causing signature mismatch
+	fee := txOpts.Fee
+	if fee == nil {
+		fee = big.NewInt(0)
+	}
+
 	// Build unsigned transaction
 	tx := &types.Transaction{
 		Body: &types.TransactionBody{
 			Payload:     payloadBytes,
 			PayloadType: payload.Type(),
-			Fee:         txOpts.Fee,
+			Fee:         fee,
 			Nonce:       uint64(txOpts.Nonce),
 			ChainID:     chainID,
 		},
+		Serialization: types.DefaultSignedMsgSerType, // Required for EthPersonalSigner
 	}
 
 	// Sign transaction
 	if err := tx.Sign(t.signer); err != nil {
 		return types.Hash{}, fmt.Errorf("failed to sign transaction: %w", err)
 	}
 
-	// Broadcast transaction
-	params := map[string]any{
-		"tx": tx,
+	// CRITICAL WORKAROUND: Pre-serialize transaction to avoid WASM pointer corruption
+	// Go WASM uses 64-bit pointers but WASM runtime uses 32-bit pointers.
+	// Transaction struct contains pointer fields (Signature, Body) which get
+	// corrupted when crossing the WASM boundary (golang/go#59156, golang/go#66984).
+	// Solution: Manually construct JSON-RPC request to avoid struct traversal in WASM.
+	txJSON, err := json.Marshal(tx)
+	if err != nil {
+		return types.Hash{}, fmt.Errorf("failed to marshal transaction: %w", err)
 	}
 
+	// Manually construct JSON-RPC request to bypass params map
+	reqID := t.nextReqID()
+	rpcReqJSON := fmt.Sprintf(
+		`{"jsonrpc":"2.0","id":"%s","method":"user.broadcast","params":{"tx":%s}}`,
+		reqID, string(txJSON))
+
+	// Create headers
+	headers := map[string]string{
+		"Content-Type": "application/json",
+	}
+
+	// Add auth cookie if we have one
+	t.authCookieMu.RLock()
+	if t.authCookie != "" {
+		headers["Cookie"] = t.authCookie
+	}
+	t.authCookieMu.RUnlock()
+
+	// Create CRE HTTP request
+	httpReq := &http.Request{
+		Url:     t.endpoint,
+		Method:  "POST",
+		Body:    []byte(rpcReqJSON),
+		Headers: headers,
+	}
+
+	// Execute via CRE client
+	httpResp, err := t.client.SendRequest(t.runtime, httpReq).Await()
+	if err != nil {
+		return types.Hash{}, fmt.Errorf("CRE HTTP request failed: %w", err)
+	}
+
+	// Check HTTP status
+	if httpResp.StatusCode != 200 {
+		return types.Hash{}, fmt.Errorf("unexpected HTTP status code: %d", httpResp.StatusCode)
+	}
+
+	// Parse JSON-RPC response
+	var rpcResp jsonrpc.Response
+	if err := json.Unmarshal(httpResp.Body, &rpcResp); err != nil {
+		return types.Hash{}, fmt.Errorf("failed to unmarshal JSON-RPC response: %w", err)
+	}
+
+	// Check for JSON-RPC errors
+	if rpcResp.Error != nil {
+		// For broadcast errors (-201), decode the BroadcastError details
+		if rpcResp.Error.Code == -201 && len(rpcResp.Error.Data) > 0 {
+			var broadcastErr struct {
+				Code    uint32 `json:"code"`
+				Hash    string `json:"hash"`
+				Message string `json:"message"`
+			}
+			if err := json.Unmarshal(rpcResp.Error.Data, &broadcastErr); err == nil {
+				return types.Hash{}, fmt.Errorf("JSON-RPC error: %s (code: %d) [Broadcast: code=%d, hash=%s, msg=%s]",
+					rpcResp.Error.Message, rpcResp.Error.Code,
+					broadcastErr.Code, broadcastErr.Hash, broadcastErr.Message)
+			}
+		}
+		return types.Hash{}, fmt.Errorf("JSON-RPC error: %s (code: %d)", rpcResp.Error.Message, rpcResp.Error.Code)
+	}
+
+	// Unmarshal result
 	var result struct {
 		TxHash types.Hash `json:"tx_hash"`
 	}
-
-	if err := t.callJSONRPC(ctx, "user.broadcast", params, &result); err != nil {
-		return types.Hash{}, err
+	if err := json.Unmarshal(rpcResp.Result, &result); err != nil {
+		return types.Hash{}, fmt.Errorf("failed to unmarshal result: %w", err)
 	}
 
 	return result.TxHash, nil
