didplcbft / abciapp / execution.go

package abciapp

import (
	"bytes"
	"context"
	"fmt"
	"slices"
	"strings"
	"time"

	abcitypes "github.com/cometbft/cometbft/abci/types"
	cbornode "github.com/ipfs/go-ipld-cbor"
	"github.com/palantir/stacktrace"
	"github.com/samber/lo"
)

// InitChain implements [types.Application].
func (d *DIDPLCApplication) InitChain(context.Context, *abcitypes.RequestInitChain) (*abcitypes.ResponseInitChain, error) {
	// TODO
	return &abcitypes.ResponseInitChain{}, nil
}

// PrepareProposal implements [types.Application].
func (d *DIDPLCApplication) PrepareProposal(ctx context.Context, req *abcitypes.RequestPrepareProposal) (*abcitypes.ResponsePrepareProposal, error) {
	defer d.DiscardChanges()

	if req.Height == 2 {
		tx := Transaction[SetAuthoritativePlcArguments]{
			Action: TransactionActionSetAuthoritativePlc,
			Arguments: SetAuthoritativePlcArguments{
				PLCURL:        "https://plc.directory",
				RestartImport: true,
			},
		}

		out, err := cbornode.DumpObject(tx)
		if err != nil {
			return nil, stacktrace.Propagate(err, "")
		}

		req.Txs = append(req.Txs, out)
	}

	st := time.Now()
	acceptedTx := make([][]byte, 0, len(req.Txs))
	toProcess := req.Txs
	for {
		toTryNext := [][]byte{}
		for _, tx := range toProcess {
			result, err := processTx(ctx, d.transactionProcessorDependencies(), tx, req.Time, true)
			if err != nil {
				return nil, stacktrace.Propagate(err, "")
			}

			if result.Code == 0 {
				acceptedTx = append(acceptedTx, tx)
			} else {
				// if a transaction is invalid, it _might_ be because it depends on a transaction that's further up in the list
				// process it after all the others
				toTryNext = append(toTryNext, tx)
			}
		}
		if len(toProcess) >= len(toTryNext) {
			// we made no progress in this iteration - all transactions left to process fail to do so
			// so they can't be depending on anything that would be included in this block, at this point
			// just continue while dropping the transactions that would never succeed in this block
			break
		}
		if time.Since(st) > 800*time.Millisecond {
			// this is taking too long, just continue with what's already in acceptedTx
			break
		}
		toProcess = toTryNext
	}

	maybeTx, err := d.maybeCreateAuthoritativeImportTx(ctx)
	if err != nil {
		// TODO don't fail absolutely silently always, we should at least check what the error is
		//return nil, stacktrace.Propagate(err, "")
	}

	if err == nil && len(maybeTx) != 0 {
		totalSize := lo.SumBy(acceptedTx, func(tx []byte) int { return len(tx) })
		// 4 KB safety margin
		if totalSize+len(maybeTx) < int(req.MaxTxBytes)-4096 {
			// we have space to fit the import transaction

			// set execute to false to save a lot of time
			// (we trust that running the import will succeed, so just do bare minimum checks here)
			result, err := processTx(ctx, d.transactionProcessorDependencies(), maybeTx, req.Time, false)
			if err != nil {
				return nil, stacktrace.Propagate(err, "")
			}
			if result.Code == 0 {
				acceptedTx = append(acceptedTx, maybeTx)
			}
		}
	}

	return &abcitypes.ResponsePrepareProposal{Txs: acceptedTx}, nil
}

// ProcessProposal implements [types.Application].
func (d *DIDPLCApplication) ProcessProposal(ctx context.Context, req *abcitypes.RequestProcessProposal) (*abcitypes.ResponseProcessProposal, error) {
	// do not rollback tree in this method, in case the changes can be reused in FinalizeBlock
	if req.Height != d.tree.WorkingVersion() {
		// our tree went out of sync, this should never happen
		return &abcitypes.ResponseProcessProposal{Status: abcitypes.ResponseProcessProposal_REJECT}, nil
	}

	// if we return early, ensure we don't use incomplete results where we haven't voted ACCEPT
	d.lastProcessedProposalHash = nil
	d.lastProcessedProposalExecTxResults = nil
	defer func() {
		if d.lastProcessedProposalHash == nil {
			// we didn't vote ACCEPT
			// we could rollback only eventually on FinalizeBlock, but why wait - rollback now for safety
			d.DiscardChanges()
		}
	}()

	txResults := make([]*processResult, len(req.Txs))
	for i, tx := range req.Txs {
		result, action, processor, err := beginProcessTx(tx)
		if err != nil {
			return nil, stacktrace.Propagate(err, "")
		}
		if result.Code == 0 {
			if action == TransactionActionAuthoritativeImport && i != len(req.Txs)-1 {
				// if an Authoritative Import transaction is present on the block, it must be the last one
				return &abcitypes.ResponseProcessProposal{Status: abcitypes.ResponseProcessProposal_REJECT}, nil
			}

			st := time.Now()
			result, err = finishProcessTx(ctx, d.transactionProcessorDependencies(), processor, tx, req.Time, true)
			if err != nil {
				return nil, stacktrace.Propagate(err, "")
			}
			fmt.Println("FINISHPROCESSTX TOOK", time.Since(st))
		}

		// when preparing a proposal, invalid transactions should have been discarded
		// so, if something doesn't succeed now, something has gone wrong and we should not vote in agreement of the proposal
		if result.Code != 0 {
			return &abcitypes.ResponseProcessProposal{Status: abcitypes.ResponseProcessProposal_REJECT}, nil
		}

		txResults[i] = result
	}

	d.lastProcessedProposalHash = slices.Clone(req.Hash)
	d.lastProcessedProposalExecTxResults = txResults

	return &abcitypes.ResponseProcessProposal{Status: abcitypes.ResponseProcessProposal_ACCEPT}, nil
}

// ExtendVote implements [types.Application].
func (d *DIDPLCApplication) ExtendVote(context.Context, *abcitypes.RequestExtendVote) (*abcitypes.ResponseExtendVote, error) {
	// TODO
	return &abcitypes.ResponseExtendVote{}, nil
}

// VerifyVoteExtension implements [types.Application].
func (d *DIDPLCApplication) VerifyVoteExtension(context.Context, *abcitypes.RequestVerifyVoteExtension) (*abcitypes.ResponseVerifyVoteExtension, error) {
	// TODO
	return &abcitypes.ResponseVerifyVoteExtension{}, nil
}

// FinalizeBlock implements [types.Application].
func (d *DIDPLCApplication) FinalizeBlock(ctx context.Context, req *abcitypes.RequestFinalizeBlock) (*abcitypes.ResponseFinalizeBlock, error) {
	if bytes.Equal(req.Hash, d.lastProcessedProposalHash) && d.lastProcessedProposalExecTxResults != nil {
		// the block that was decided was the one we processed in ProcessProposal, and ProcessProposal processed successfully
		// reuse the uncommitted results
		return &abcitypes.ResponseFinalizeBlock{
			TxResults: lo.Map(d.lastProcessedProposalExecTxResults, func(result *processResult, _ int) *abcitypes.ExecTxResult {
				return result.ToABCI()
			}),
			AppHash: d.tree.WorkingHash(),
		}, nil
	}
	// a block other than the one we processed in ProcessProposal was decided
	// discard the current modified state, and process the decided block
	d.DiscardChanges()

	txResults := make([]*processResult, len(req.Txs))
	for i, tx := range req.Txs {
		var err error
		txResults[i], err = processTx(ctx, d.transactionProcessorDependencies(), tx, req.Time, true)
		if err != nil {
			return nil, stacktrace.Propagate(err, "")
		}
	}

	d.lastProcessedProposalHash = slices.Clone(req.Hash)
	d.lastProcessedProposalExecTxResults = txResults

	return &abcitypes.ResponseFinalizeBlock{
		TxResults: lo.Map(d.lastProcessedProposalExecTxResults, func(result *processResult, _ int) *abcitypes.ExecTxResult {
			return result.ToABCI()
		}),
		AppHash: d.tree.WorkingHash(),
	}, nil
}

// Commit implements [types.Application].
func (d *DIDPLCApplication) Commit(context.Context, *abcitypes.RequestCommit) (*abcitypes.ResponseCommit, error) {
	_, newVersion, err := d.tree.SaveVersion()
	if err != nil {
		return nil, stacktrace.Propagate(err, "")
	}

	for _, r := range d.lastProcessedProposalExecTxResults {
		for _, cb := range r.commitSideEffects {
			cb()
		}
	}

	minHeightToKeep := max(newVersion-5, 0)
	minVerToKeep := max(minHeightToKeep-5, 0)
	if minVerToKeep > 0 {
		err = d.tree.DeleteVersionsTo(minVerToKeep)
		if err != nil && !strings.Contains(err.Error(), "active readers") {
			return nil, stacktrace.Propagate(err, "")
		}
	}

	return &abcitypes.ResponseCommit{
		// TODO only discard actual blockchain history based on settings
		//RetainHeight: minHeightToKeep,
	}, nil
}

func (d *DIDPLCApplication) transactionProcessorDependencies() TransactionProcessorDependencies {
	return TransactionProcessorDependencies{
		runnerContext: d.runnerContext,
		plc:           d.plc,
		tree:          d,
		aocsByPLC:     d.aocsByPLC,
	}
}