Skip to content

fix in progress logs request #800

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Unheilbar merged 5 commits into from
Oct 5, 2026
Merged

fix in progress logs request #800

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
38ce9eb
fix in progress logs request
Unheilbar Sep 24, 2026
8cefe13
fix typo
Unheilbar Sep 24, 2026
7f970cd
register processed filter
Unheilbar Sep 29, 2026
3f61226
derive signature from logs for successful tx
Unheilbar Oct 2, 2026
2bcac9a
limit in progress logs to 1
Unheilbar Oct 5, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
191 changes: 65 additions & 126 deletions chain_capabilities/solana/actions/transmission_info_provider.go
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,44 +1,42 @@
package actions

import (
"bytes"
"context"
"encoding/base64"
"encoding/json"
"errors"
"fmt"
"strings"

"github.com/gagliardetto/solana-go"
"github.com/gagliardetto/solana-go/rpc"

"github.com/smartcontractkit/chainlink-common/pkg/types"
soltypes "github.com/smartcontractkit/chainlink-common/pkg/types/chains/solana"
"github.com/smartcontractkit/chainlink-common/pkg/types/query"
"github.com/smartcontractkit/chainlink-common/pkg/types/query/primitives"
solprimitives "github.com/smartcontractkit/chainlink-common/pkg/types/query/primitives/solana"
"github.com/smartcontractkit/chainlink-solana/contracts"
ks_forwarder "github.com/smartcontractkit/chainlink-solana/contracts/generated/keystone_forwarder"
lptypes "github.com/smartcontractkit/chainlink-solana/pkg/solana/logpoller/types"
)

const eventReportInProgress = "ReportInProgress"
const eventReportProcessed = "ReportProcessed"

// transmissionLogSubkeyPath indexes ReportInProgress by transmission_id.
// transmissionLogSubkeyPath indexes forwarder events by transmission_id.
var transmissionLogSubkeyPath = []string{"TransmissionId"}

// forwarderStateSubkeyPath indexes ReportInProgress by forwarder_state.
// stateSubkeyPath indexes forwarder events by forwarder_state.
var stateSubkeyPath = []string{"State"}

type logReader struct {
types.SolanaService
forwarderProgramID solana.PublicKey
forwarderState solana.PublicKey
sigInProgress soltypes.EventSignature
sigProcessed soltypes.EventSignature
}

// OnChainTransmissionInfoProvider uses the ExecutionState PDA for success/failure and
// ReportInProgress logs for transaction signatures (ReportProcessed is not used; it can be truncated).
// OnChainTransmissionInfoProvider derives transmission state from tracked forwarder logs.
// A ReportProcessed log is tracked only from successfully executed transactions, so its
// presence proves success and its tx hash is always a successful tx signature.
// ReportInProgress logs are tracked including reverted transactions: they prove an attempt
// was made and, absent a ReportProcessed log, that every attempt so far failed.
type OnChainTransmissionInfoProvider struct {
types.SolanaService
forwarderProgramID solana.PublicKey
Expand All @@ -58,6 +56,9 @@ func newOnChainTransmissionInfoProvider(ctx context.Context, programID, forwarde
if err := lr.unregisterLegacyInProgressFilter(ctx); err != nil {
return nil, fmt.Errorf("failed to unregister legacy ReportInProgress log filter: %w", err)
}
if err := lr.registerProcessedFilter(ctx); err != nil {
return nil, fmt.Errorf("failed to register ReportProcessed log filter: %w", err)
}
return &OnChainTransmissionInfoProvider{
SolanaService: s,
forwarderProgramID: programID,
Expand All @@ -71,109 +72,27 @@ func (p *OnChainTransmissionInfoProvider) GetTransmissionInfo(ctx context.Contex
if err != nil {
return TransmissionInfo{}, fmt.Errorf("failed to request ReportInProgress events: %w", err)
}

if len(inProgressLogs) == 0 {
return TransmissionInfo{State: TransmissionStateNotAttempted}, nil
}

execStateAddr, err := deriveExecutionStatePDA(p.forwarderState, transmissionID, p.forwarderProgramID)
if err != nil {
return TransmissionInfo{}, fmt.Errorf("failed to derive execution state PDA: %w", err)
}

reply, err := p.GetAccountInfoWithOpts(ctx, soltypes.GetAccountInfoRequest{
Account: soltypes.PublicKey(execStateAddr),
Opts: &soltypes.GetAccountInfoOpts{
Commitment: soltypes.CommitmentProcessed,
},
})
if err != nil {
if !isExecutionStateAccountMissing(err) {
return TransmissionInfo{}, fmt.Errorf("failed to get execution state account: %w", err)
}
reply = &soltypes.GetAccountInfoReply{}
}

sig, sigErr := signatureFromInProgressLogs(inProgressLogs)
if sigErr != nil {
return TransmissionInfo{}, sigErr
}

raw, haveBinary := accountDataBytesForTransmission(reply)
if !haveBinary {
// ReportInProgress but no decodable account payload (reverted tx, or missing data on wire).
return TransmissionInfo{State: TransmissionStateFailed, Signature: sig}, nil
}

execState, err := ks_forwarder.ParseAccount_ExecutionState(raw)
// The ReportProcessed filter excludes reverted transactions, so any tracked
// ReportProcessed log comes from a successfully executed tx.
processedLogs, err := p.lr.queryProcessed(ctx, transmissionID)
Comment thread
dhaidashenko marked this conversation as resolved.
if err != nil {
return TransmissionInfo{}, fmt.Errorf("failed to parse execution state account: %w", err)
}

if !bytes.Equal(execState.TransmissionId[:], transmissionID[:]) {
return TransmissionInfo{}, fmt.Errorf("execution state transmission id mismatch")
return TransmissionInfo{}, fmt.Errorf("failed to request ReportProcessed events: %w", err)
}

var state TransmissionState
if execState.Success {
state = TransmissionStateSucceeded
} else {
state = TransmissionStateFailed
if len(processedLogs) > 0 {
return TransmissionInfo{
State: TransmissionStateSucceeded,
Signature: solana.Signature(processedLogs[0].TxHash),
}, nil
}

return TransmissionInfo{
State: state,
Signature: sig,
}, nil
}

// accountDataBytesForTransmission returns raw program data for Anchor parsing. Prefers
// AsDecodedBinary; if empty (e.g. jsonParsed-only path), decodes Solana's ["base64","base64"] from AsJSON.
func accountDataBytesForTransmission(reply *soltypes.GetAccountInfoReply) ([]byte, bool) {
if reply == nil || reply.Value == nil || reply.Value.Data == nil {
return nil, false
}
d := reply.Value.Data
if len(d.AsDecodedBinary) > 0 {
return d.AsDecodedBinary, true
}
raw, err := accountDataBytesFromJSON(d.AsJSON)
if err != nil || len(raw) == 0 {
return nil, false
}
return raw, true
}

func accountDataBytesFromJSON(asJSON []byte) ([]byte, error) {
if len(asJSON) == 0 {
return nil, fmt.Errorf("empty account data json")
}
var arr []string
if err := json.Unmarshal(asJSON, &arr); err == nil && len(arr) >= 2 && arr[1] == "base64" {
return base64.StdEncoding.DecodeString(arr[0])
}
var wrapped struct {
Data json.RawMessage `json:"data"`
}
if err := json.Unmarshal(asJSON, &wrapped); err == nil && len(wrapped.Data) > 0 {
return accountDataBytesFromJSON(wrapped.Data)
}
return nil, fmt.Errorf("could not extract base64 account data from json")
}

func signatureFromInProgressLogs(inProgressLogs []*soltypes.Log) (solana.Signature, error) {
if len(inProgressLogs) == 0 {
return solana.Signature{}, fmt.Errorf("no in-progress logs")
}
log := inProgressLogs[0]
minBlock := inProgressLogs[0].BlockNumber
for _, l := range inProgressLogs {
if l.BlockNumber < minBlock {
log = l
minBlock = l.BlockNumber
}
}
return solana.Signature(log.TxHash), nil
// ReportInProgress without ReportProcessed: the report was attempted, but every
// attempt so far landed in a reverted tx (a successful tx would have emitted a
// tracked ReportProcessed log).
return TransmissionInfo{State: TransmissionStateFailed, Signature: solana.Signature(inProgressLogs[0].TxHash)}, nil
}

// Legacy LogTracking doesn't validate against forwarder state used in Event.
Expand Down Expand Up @@ -202,8 +121,32 @@ func (lr *logReader) registerInProgressFilter(ctx context.Context) error {
return nil
}

func (lr *logReader) registerProcessedFilter(ctx context.Context) error {
idlJSON := []byte(contracts.FetchForwarderIDL())
sigProcessed := soltypes.EventSignature(lptypes.NewEventSignatureFromName(eventReportProcessed))
err := lr.RegisterLogTracking(ctx, soltypes.LPFilterQuery{
Name: eventReportProcessed + "_" + lr.forwarderProgramID.String() + "_v2",
Address: soltypes.PublicKey(lr.forwarderProgramID),
EventName: eventReportProcessed,
EventSig: sigProcessed,
ContractIdlJSON: idlJSON,
SubkeyPaths: [][]string{transmissionLogSubkeyPath, stateSubkeyPath},
// Only successfully executed transactions are tracked, so signatures queried
// through this filter always belong to a successful tx.
IncludeReverted: false,
})
if err != nil {
return fmt.Errorf("failed to register ReportProcessed filter for forwarder: %w", err)
}

lr.sigProcessed = sigProcessed
return nil
}

const inProgressLogsLimit = 1

func (lr *logReader) queryInProgress(ctx context.Context, transmissionID [32]byte) ([]*soltypes.Log, error) {
limit := query.NewLimitAndSort(query.CountLimit(1), query.NewSortBySequence(query.Desc))
limit := query.NewLimitAndSort(query.CountLimit(inProgressLogsLimit), query.NewSortBySequence(query.Asc))
exprs := []query.Expression{
solprimitives.NewEventSigFilter(lr.sigInProgress),
solprimitives.NewAddressFilter(soltypes.PublicKey(lr.forwarderProgramID)),
Expand All @@ -223,26 +166,22 @@ func (lr *logReader) queryInProgress(ctx context.Context, transmissionID [32]byt
return logs, nil
}

func deriveExecutionStatePDA(forwarderState solana.PublicKey, transmissionID [32]byte, programID solana.PublicKey) (solana.PublicKey, error) {
seeds := [][]byte{
[]byte("execution_state"),
forwarderState.Bytes(),
transmissionID[:],
func (lr *logReader) queryProcessed(ctx context.Context, transmissionID [32]byte) ([]*soltypes.Log, error) {
limit := query.NewLimitAndSort(query.CountLimit(1), query.NewSortBySequence(query.Asc))
exprs := []query.Expression{
solprimitives.NewEventSigFilter(lr.sigProcessed),
solprimitives.NewAddressFilter(soltypes.PublicKey(lr.forwarderProgramID)),
solprimitives.NewEventBySubkeyFilter(0, []solprimitives.IndexedValueComparator{
{Value: transmissionID[:], Operator: primitives.Eq},
}),
solprimitives.NewEventBySubkeyFilter(1, []solprimitives.IndexedValueComparator{
{Value: lr.forwarderState.Bytes(), Operator: primitives.Eq},
}),
}
ret, _, err := solana.FindProgramAddress(seeds, programID)
return ret, err
}

func isExecutionStateAccountMissing(err error) bool {
if err == nil {
return false
}
if errors.Is(err, rpc.ErrNotFound) {
return true
}
s := strings.ToLower(err.Error())
if !strings.Contains(s, "not found") {
return false
logs, err := lr.QueryTrackedLogs(ctx, exprs, limit)
if err != nil {
return nil, fmt.Errorf("failed to query tracked logs: %w", err)
}
return strings.Contains(s, "account info") || strings.Contains(s, "getaccountinfo")
return logs, nil
}
Loading
Loading