Continue work on authoritative import

abciapp/execution.go

···
 
 	abcitypes "github.com/cometbft/cometbft/abci/types"
 	"github.com/palantir/stacktrace"
+	"github.com/samber/lo"
 )
 
 // InitChain implements [types.Application].
···
 // PrepareProposal implements [types.Application].
 func (d *DIDPLCApplication) PrepareProposal(ctx context.Context, req *abcitypes.RequestPrepareProposal) (*abcitypes.ResponsePrepareProposal, error) {
 	defer d.tree.Rollback()
+
+	deps := TransactionProcessorDependencies{
+		plc:  d.plc,
+		tree: d,
+	}
 
 	st := time.Now()
 	acceptedTx := make([][]byte, 0, len(req.Txs))
···
 	for {
 		toTryNext := [][]byte{}
 		for _, tx := range toProcess {
-			result, err := processTx(ctx, d.plc, tx, req.Time, true)
+			result, err := processTx(ctx, deps, tx, req.Time, true)
 			if err != nil {
 				return nil, stacktrace.Propagate(err, "")
 			}
 
+			if result.IsAuthoritativeImportTransaction {
+				// this type of transaction is not meant to appear in the mempool,
+				// but maybe it's not impossible that a non-compliant node could have gossiped it to us?
+				// (not sure if CometBFT checks transactions coming from other peers against CheckTx)
+				continue
+			}
+
 			if result.Code == 0 {
 				acceptedTx = append(acceptedTx, tx)
 			} else {
···
 				toTryNext = append(toTryNext, tx)
 			}
 		}
-		if len(toProcess) == len(toTryNext) {
+		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
···
 		toProcess = toTryNext
 	}
 
+	totalSize := lo.SumBy(acceptedTx, func(tx []byte) int { return len(tx) })
+	if totalSize < int(req.MaxTxBytes)-4096 {
+		// we have space to fit an import transaction
+		// TODO
+	}
+
 	return &abcitypes.ResponsePrepareProposal{Txs: acceptedTx}, nil
 }
 
···
 		}
 	}()
 
+	deps := TransactionProcessorDependencies{
+		plc:  d.plc,
+		tree: d,
+	}
+
 	txResults := make([]*abcitypes.ExecTxResult, len(req.Txs))
 	for i, tx := range req.Txs {
-		result, err := processTx(ctx, d.plc, tx, req.Time, true)
+		result, err := processTx(ctx, deps, tx, req.Time, true)
 		if err != nil {
 			return nil, stacktrace.Propagate(err, "")
 		}
···
 			return &abcitypes.ResponseProcessProposal{Status: abcitypes.ResponseProcessProposal_REJECT}, nil
 		}
 
+		if result.IsAuthoritativeImportTransaction && 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
+		}
+
 		txResults[i] = &abcitypes.ExecTxResult{
 			Code:      result.Code,
 			Data:      result.Data,
···
 	// discard the current modified state, and process the decided block
 	d.tree.Rollback()
 
+	deps := TransactionProcessorDependencies{
+		plc:  d.plc,
+		tree: d,
+	}
+
 	txResults := make([]*abcitypes.ExecTxResult, len(req.Txs))
 	for i, tx := range req.Txs {
-		result, err := processTx(ctx, d.plc, tx, req.Time, true)
+		result, err := processTx(ctx, deps, tx, req.Time, true)
 		if err != nil {
 			return nil, stacktrace.Propagate(err, "")
 		}

abciapp/import.go

···
+package abciapp
+
+import (
+	"bufio"
+	"context"
+	"crypto/sha256"
+	"encoding/binary"
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"net/url"
+	"time"
+
+	"github.com/bluesky-social/indigo/atproto/syntax"
+	"github.com/did-method-plc/go-didplc"
+	"github.com/ipfs/go-cid"
+	"github.com/palantir/stacktrace"
+	"github.com/samber/lo"
+	"tangled.org/gbl08ma/didplcbft/store"
+)
+
+func fetchExportedBatchFromAuthoritativeSource(ctx context.Context, plcURL string, startAt, maxCount uint64) ([]didplc.LogEntry, uint64, error) {
+	baseURL, err := url.JoinPath(plcURL, "/export")
+	if err != nil {
+		return nil, 0, stacktrace.Propagate(err, "")
+	}
+
+	client := &http.Client{Timeout: 30 * time.Second}
+
+	entries := make([]didplc.LogEntry, 0, maxCount)
+	for {
+		req, err := http.NewRequestWithContext(ctx, "GET", baseURL, nil)
+		if err != nil {
+			return nil, 0, stacktrace.Propagate(err, "")
+		}
+
+		req.Header.Set("User-Agent", "go-did-method-plc")
+
+		requestCount := min(1000, maxCount-uint64(len(entries)))
+
+		q := req.URL.Query()
+		q.Add("count", fmt.Sprint(requestCount))
+		q.Add("after", fmt.Sprint(startAt))
+		req.URL.RawQuery = q.Encode()
+
+		resp, err := client.Do(req)
+		if err != nil {
+			return nil, 0, stacktrace.Propagate(err, "")
+		}
+		defer resp.Body.Close()
+
+		if resp.StatusCode != http.StatusOK {
+			return nil, 0, stacktrace.NewError("non-200 status code")
+		}
+
+		type logEntryWithSeq struct {
+			didplc.LogEntry
+			Seq uint64 `json:"seq"`
+		}
+
+		// Read response body
+		s := bufio.NewScanner(resp.Body)
+		numEntriesThisResponse := 0
+		for s.Scan() && len(entries) < int(maxCount) {
+			var entry logEntryWithSeq
+			if err := json.Unmarshal(s.Bytes(), &entry); err != nil {
+				return nil, 0, stacktrace.Propagate(err, "")
+			}
+			entries = append(entries, entry.LogEntry)
+			numEntriesThisResponse++
+			startAt = entry.Seq
+		}
+		if s.Err() != nil {
+			return nil, 0, stacktrace.Propagate(s.Err(), "")
+		}
+
+		if uint64(numEntriesThisResponse) < requestCount || len(entries) >= int(maxCount) {
+			break
+		}
+	}
+
+	return entries, startAt, nil
+}
+
+func computeLogEntriesHash(logEntries []didplc.LogEntry) ([]byte, error) {
+	// let's _not_ rely on the specifics of the JSON representation
+	// (instead let's rely on specifics of our implementation, heh)
+
+	hash := sha256.New()
+	for i, entry := range logEntries {
+		// Write DID
+		didBytes, err := store.DIDToBytes(entry.DID)
+		if err != nil {
+			return nil, stacktrace.Propagate(err, "invalid DID in entry index %d", i)
+		}
+
+		_, err = hash.Write(didBytes)
+		if err != nil {
+			return nil, stacktrace.Propagate(err, "")
+		}
+
+		// Write CID
+		// (We trust that the authoritative source computed CIDs properly, so we use theirs rather than decoding the operation and recomputing)
+		cid, err := cid.Decode(entry.CID)
+		if err != nil {
+			return nil, stacktrace.Propagate(err, "invalid CID in entry index %d", i)
+		}
+
+		_, err = hash.Write(cid.Bytes())
+		if err != nil {
+			return nil, stacktrace.Propagate(err, "")
+		}
+
+		// Write CreatedAt
+		createdAt, err := syntax.ParseDatetime(entry.CreatedAt)
+		if err != nil {
+			return nil, stacktrace.Propagate(err, "invalid CreatedAt in entry index %d", i)
+		}
+
+		tsBytes := make([]byte, 8)
+		binary.BigEndian.PutUint64(tsBytes, uint64(createdAt.Time().Truncate(1*time.Millisecond).UTC().UnixNano()))
+
+		_, err = hash.Write(tsBytes)
+		if err != nil {
+			return nil, stacktrace.Propagate(err, "")
+		}
+
+		// Write Nullified
+		_, err = hash.Write([]byte{lo.Ternary[byte](entry.Nullified, 1, 0)})
+		if err != nil {
+			return nil, stacktrace.Propagate(err, "")
+		}
+	}
+
+	return hash.Sum(nil), nil
+}

abciapp/mempool.go

···
 
 // CheckTx implements [types.Application].
 func (d *DIDPLCApplication) CheckTx(ctx context.Context, req *abcitypes.RequestCheckTx) (*abcitypes.ResponseCheckTx, error) {
-	result, err := processTx(ctx, d.plc, req.Tx, time.Now(), false)
+	deps := TransactionProcessorDependencies{
+		plc:  d.plc,
+		tree: d,
+	}
+
+	result, err := processTx(ctx, deps, req.Tx, time.Now(), false)
 	if err != nil {
 		return nil, stacktrace.Propagate(err, "")
+	}
+	if result.IsAuthoritativeImportTransaction {
+		// this type of transaction is meant to be included only by validator nodes
+		return &abcitypes.ResponseCheckTx{
+			Code: 4002,
+			Info: "AuthoritativeImport transactions can only be introduced by validator nodes",
+		}, nil
 	}
 	return &abcitypes.ResponseCheckTx{
 		Code:      result.Code,

abciapp/tx.go

···
 
 type TransactionAction string
 
-var (
-	knownActions                         = map[TransactionAction]struct{}{}
-	TransactionActionCreatePlcOp         = registerTransactionAction[CreatePlcOpArguments]("CreatePlcOp")
-	TransactionActionAuthoritativeImport = registerTransactionAction[AuthoritativeImportArguments]("AuthoritativeImport")
-)
+type TransactionProcessorDependencies struct {
+	plc  plc.PLC
+	tree plc.TreeProvider // TODO maybe we should move the TreeProvider definition out of the plc package then?
+}
+
+type TransactionProcessor func(ctx context.Context, deps TransactionProcessorDependencies, txBytes []byte, atTime time.Time, execute bool) (*processResult, error)
+
+var knownActions = map[TransactionAction]TransactionProcessor{}
 
-func registerTransactionAction[ArgType ArgumentType](action string) TransactionAction {
+func registerTransactionAction[ArgType ArgumentType](action string, processor TransactionProcessor) TransactionAction {
 	ta := TransactionAction(action)
 	if _, present := knownActions[ta]; present {
 		panic("action already registered")
···
 	if argType.ForAction() != "" && argType.ForAction() != ta {
 		panic("mismatched argument types")
 	}
-	knownActions[ta] = struct{}{}
+	knownActions[ta] = processor
 	return ta
 }
 
···
 }
 
 type processResult struct {
+	IsAuthoritativeImportTransaction bool
+
 	Code      uint32
 	Data      []byte
 	Log       string
···
 	Codespace string
 }
 
-func processTx(ctx context.Context, p plc.PLC, txBytes []byte, atTime time.Time, execute bool) (*processResult, error) {
+func processTx(ctx context.Context, deps TransactionProcessorDependencies, txBytes []byte, atTime time.Time, execute bool) (*processResult, error) {
 	if !IsTransactionSanitized(txBytes) {
 		return &processResult{
 			Code: 4000,
···
 		}, nil
 	}
 
-	var result *processResult
-	switch TransactionAction(action) {
-	case TransactionActionCreatePlcOp:
-		result, err = processCreatePlcOpTx(ctx, p, txBytes, atTime, execute)
-	case TransactionActionAuthoritativeImport:
-		result, err = processAuthoritativeImportTx(ctx, p, txBytes, atTime, execute)
-	default:
-		result = &processResult{
+	processor, ok := knownActions[TransactionAction(action)]
+	if !ok {
+		return &processResult{
 			Code: 4001,
 			Info: "Unknown transaction action",
-		}
-
+		}, nil
 	}
+
+	result, err := processor(ctx, deps, txBytes, atTime, execute)
 	return result, stacktrace.Propagate(err, "")
 }

abciapp/tx_create_plc_op.go

···
 	"tangled.org/gbl08ma/didplcbft/plc"
 )
 
+var TransactionActionCreatePlcOp = registerTransactionAction[CreatePlcOpArguments]("CreatePlcOp", processCreatePlcOpTx)
+
 type CreatePlcOpArguments struct {
 	DID       string         `json:"did" refmt:"did"`
 	Operation *didplc.OpEnum `refmt:"operation"`
···
 	cbornode.RegisterCborType(Transaction[CreatePlcOpArguments]{})
 }
 
-func processCreatePlcOpTx(ctx context.Context, p plc.PLC, txBytes []byte, atTime time.Time, execute bool) (*processResult, error) {
+func processCreatePlcOpTx(ctx context.Context, deps TransactionProcessorDependencies, txBytes []byte, atTime time.Time, execute bool) (*processResult, error) {
 	tx, err := UnmarshalTransaction[CreatePlcOpArguments](txBytes)
 	if err != nil {
 		return &processResult{
···
 	}
 
 	if execute {
-		err = p.ExecuteOperation(ctx, atTime, tx.Arguments.DID, opBytes)
+		err = deps.plc.ExecuteOperation(ctx, atTime, tx.Arguments.DID, opBytes)
 	} else {
-		err = p.ValidateOperation(ctx, plc.CommittedTreeVersion, atTime, tx.Arguments.DID, opBytes)
+		err = deps.plc.ValidateOperation(ctx, plc.CommittedTreeVersion, atTime, tx.Arguments.DID, opBytes)
 	}
 	if err != nil {
 		if code, ok := plc.InvalidOperationErrorCode(err); ok {

abciapp/tx_import.go

···
 
 import (
 	"context"
+	"encoding/hex"
+	"net/url"
 	"time"
 
+	"github.com/did-method-plc/go-didplc"
 	cbornode "github.com/ipfs/go-ipld-cbor"
 	"github.com/palantir/stacktrace"
 	"tangled.org/gbl08ma/didplcbft/plc"
+	"tangled.org/gbl08ma/didplcbft/store"
 )
 
+var TransactionActionSetAuthoritativePlc = registerTransactionAction[SetAuthoritativePlcArguments]("SetAuthoritativePlc", processSetAuthoritativePlcTx)
+
+type SetAuthoritativePlcArguments struct {
+	PLCURL        string `json:"plcURL" refmt:"plcURL"`
+	RestartImport bool   `json:"restartImport" refmt:"restartImport"`
+}
+
+func (SetAuthoritativePlcArguments) ForAction() TransactionAction {
+	return TransactionActionSetAuthoritativePlc
+}
+
+func init() {
+	cbornode.RegisterCborType(SetAuthoritativePlcArguments{})
+	cbornode.RegisterCborType(Transaction[SetAuthoritativePlcArguments]{})
+}
+
+func processSetAuthoritativePlcTx(ctx context.Context, deps TransactionProcessorDependencies, txBytes []byte, atTime time.Time, execute bool) (*processResult, error) {
+	tx, err := UnmarshalTransaction[SetAuthoritativePlcArguments](txBytes)
+	if err != nil {
+		return &processResult{
+			Code: 4000,
+			Info: err.Error(),
+		}, nil
+	}
+
+	// TODO this transaction must somehow validate that whoever submitted it has the permission to change this
+	// Does it even make sense to keep this operation type as something submitted via the mempool in the long run,
+	// or would it be tied to some sort of proposal/participation system, where the validators submit this operation type in response to some on-chain trigger?
+
+	// A simple solution in the short term might be to just validate a "simple" public-private signature + an expiry timestamp (to prevent replay attacks)
+	// which would both be part of the SetAuthoritativePlcArguments. Very centralized, but very straightforward
+	// (the public key would be part of the config or even hardcoded for good measure)
+
+	if tx.Arguments.PLCURL != "" {
+		parsed, err := url.Parse(tx.Arguments.PLCURL)
+		if err != nil || parsed.Scheme != "https" {
+			return &processResult{
+				Code: 4100,
+				Info: "Malformed Authoritative PLC URL",
+			}, nil
+		}
+	}
+
+	if execute {
+		tree, err := deps.tree.MutableTree()
+		if err != nil {
+			return nil, stacktrace.Propagate(err, "")
+		}
+		err = store.Tree.SetAuthoritativePLC(tree, tx.Arguments.PLCURL)
+		if err != nil {
+			return nil, stacktrace.Propagate(err, "")
+		}
+
+		if tx.Arguments.RestartImport {
+			err = store.Tree.SetAuthoritativeImportProgress(tree, 0)
+			if err != nil {
+				return nil, stacktrace.Propagate(err, "")
+			}
+		}
+	}
+
+	return &processResult{
+		Code: 0,
+	}, nil
+}
+
+var TransactionActionAuthoritativeImport = registerTransactionAction[AuthoritativeImportArguments]("AuthoritativeImport", processAuthoritativeImportTx)
+
 type AuthoritativeImportArguments struct {
-	Cursor string `json:"cursor" refmt:"cursor"`
+	PLCURL string `json:"plcURL" refmt:"plcURL"`
+	Cursor uint64 `json:"cursor" refmt:"cursor"`
+	Count  uint64 `json:"count" refmt:"count"`
 	Hash   string `json:"hash" refmt:"hash"`
 }
 
···
 	cbornode.RegisterCborType(Transaction[AuthoritativeImportArguments]{})
 }
 
-func processAuthoritativeImportTx(ctx context.Context, p plc.PLC, txBytes []byte, atTime time.Time, execute bool) (*processResult, error) {
-	// TODO
-	return nil, stacktrace.NewError("not implemented")
+func processAuthoritativeImportTx(ctx context.Context, deps TransactionProcessorDependencies, txBytes []byte, atTime time.Time, execute bool) (*processResult, error) {
+	tx, err := UnmarshalTransaction[AuthoritativeImportArguments](txBytes)
+	if err != nil {
+		return &processResult{
+			Code: 4000,
+			Info: err.Error(),
+		}, nil
+	}
+
+	roTree, err := deps.tree.ImmutableTree(plc.CommittedTreeVersion)
+	if err != nil {
+		return nil, stacktrace.Propagate(err, "")
+	}
+
+	expectedPlcUrl, err := store.Tree.AuthoritativePLC(roTree)
+	if err != nil {
+		return nil, stacktrace.Propagate(err, "")
+	}
+
+	if expectedPlcUrl != tx.Arguments.PLCURL {
+		return &processResult{
+			Code: 4110,
+			Info: "Unexpected Authoritative PLC URL",
+		}, nil
+	}
+
+	expectedCursor, err := store.Tree.AuthoritativeImportProgress(roTree)
+	if err != nil {
+		return nil, stacktrace.Propagate(err, "")
+	}
+
+	if expectedCursor != tx.Arguments.Cursor {
+		return &processResult{
+			Code: 4111,
+			Info: "Unexpected import cursor",
+		}, nil
+	}
+
+	// TODO this shouldn't be happening synchronously! We should always be ahead of the next transaction!
+	// or at the very least it should only happen once (e.g. when processing the proposal) and then we should cache until it expires or until we actually commit
+	operations, newCursor, err := fetchExportedBatchFromAuthoritativeSource(ctx, expectedPlcUrl, expectedCursor, tx.Arguments.Count)
+	if err != nil {
+		// returning an actual error like this means "consensus failure". Probably not the best way to deal with this, we would rather drop the transaction if not all nodes can fetch the same thing
+		// TODO investigate
+		return nil, stacktrace.Propagate(err, "")
+	}
+
+	expectedHashBytes, err := computeLogEntriesHash(operations)
+	if err != nil {
+		return nil, stacktrace.Propagate(err, "")
+	}
+
+	if hex.EncodeToString(expectedHashBytes) != tx.Arguments.Hash {
+		return &processResult{
+			Code: 4112,
+			Info: "Unexpected import hash",
+		}, nil
+	}
+
+	if execute {
+		tree, err := deps.tree.MutableTree()
+		if err != nil {
+			return nil, stacktrace.Propagate(err, "")
+		}
+
+		var client didplc.Client
+		for _, entry := range operations {
+			err := deps.plc.ImportOperationFromAuthoritativeSource(ctx, entry, func() ([]didplc.LogEntry, error) {
+				// TODO Oh NOOOOOOO! This is not deterministic
+				// (fetched at different times, the audit log might grow, therefore we'll fetch and insert more ops, and change the apphash)
+				// we need to either limit how much audit log we return (only doable if how much was fetched for each op was part of the tx, ugh)
+				// or (probably preferred approach) make it so that the ImportOperationFromAuthoritativeSource / ReplaceHistory function only replaces up until the CID that's being imported, and no further ops
+				// Even then there is a problem: what if the nullified status changes between imports :dizzy_face:
+				// (can the nullified status change for the ops that are being imported only up until CID? Need to think)
+				e, err := client.AuditLog(ctx, entry.DID)
+				return e, stacktrace.Propagate(err, "")
+			})
+			if err != nil {
+				return nil, stacktrace.Propagate(err, "")
+			}
+		}
+		err = store.Tree.SetAuthoritativeImportProgress(tree, newCursor)
+		if err != nil {
+			return nil, stacktrace.Propagate(err, "")
+		}
+	}
+
+	// TODO finish implementation
+	// 1. if execute is true: actually import the operations
+	// 2. if execute is true: update AuthoritativeImportProgress
+
+	return &processResult{
+		IsAuthoritativeImportTransaction: true,
+		Code:                             0,
+	}, stacktrace.NewError("not implemented")
 }

store/tree.go

···
 	ExportOperations(tree ReadOnlyTree, after uint64, count int) ([]types.SequencedLogEntry, error) // passing a count of zero means unlimited
 	StoreOperation(tree *iavl.MutableTree, entry didplc.LogEntry, nullifyWithIndexEqualOrGreaterThan mo.Option[int]) error
 	ReplaceHistory(tree *iavl.MutableTree, history []didplc.LogEntry) error
+
+	AuthoritativePLC(tree ReadOnlyTree) (string, error)
+	SetAuthoritativePLC(tree *iavl.MutableTree, url string) error
+
+	AuthoritativeImportProgress(tree ReadOnlyTree) (uint64, error)
+	SetAuthoritativeImportProgress(tree *iavl.MutableTree, nextCursor uint64) error
 }
 
 var _ PLCTreeStore = (*TreeStore)(nil)
···
 func (t *TreeStore) AuditLog(tree ReadOnlyTree, did string, withProof bool) ([]types.SequencedLogEntry, *ics23.CommitmentProof, error) {
 	proofs := []*ics23.CommitmentProof{}
 
-	didBytes, err := didToBytes(did)
+	didBytes, err := DIDToBytes(did)
 	if err != nil {
 		return nil, nil, stacktrace.Propagate(err, "")
 	}
···
 
 func (t *TreeStore) AuditLogReverseIterator(tree ReadOnlyTree, did string, retErr *error) iter.Seq2[int, types.SequencedLogEntry] {
 	return func(yield func(int, types.SequencedLogEntry) bool) {
-		didBytes, err := didToBytes(did)
+		didBytes, err := DIDToBytes(did)
 		if err != nil {
 			*retErr = stacktrace.Propagate(err, "")
 			return
···
 }
 
 func (t *TreeStore) StoreOperation(tree *iavl.MutableTree, entry didplc.LogEntry, nullifyWithIndexEqualOrGreaterThan mo.Option[int]) error {
-	didBytes, err := didToBytes(entry.DID)
+	didBytes, err := DIDToBytes(entry.DID)
 	if err != nil {
 		return stacktrace.Propagate(err, "")
 	}
···
 
 	did := remoteHistory[0].DID
 
-	didBytes, err := didToBytes(did)
+	didBytes, err := DIDToBytes(did)
 	if err != nil {
 		return stacktrace.Propagate(err, "")
 	}
···
 	return seq + 1, stacktrace.Propagate(err, "")
 }
 
-func didToBytes(did string) ([]byte, error) {
+func DIDToBytes(did string) ([]byte, error) {
 	if !strings.HasPrefix(did, "did:plc:") {
 		return nil, stacktrace.NewError("invalid did:plc")
 	}
···
 			})).
 		Complete())
 }
+
+func (t *TreeStore) AuthoritativePLC(tree ReadOnlyTree) (string, error) {
+	url, err := tree.Get([]byte("aPLCURL"))
+	if err != nil {
+		return "", stacktrace.Propagate(err, "")
+	}
+	if url == nil {
+		return "", nil
+	}
+	return string(url), nil
+}
+
+func (t *TreeStore) SetAuthoritativePLC(tree *iavl.MutableTree, url string) error {
+	_, err := tree.Set([]byte("aPLCURL"), []byte(url))
+	return stacktrace.Propagate(err, "")
+}
+
+func (t *TreeStore) AuthoritativeImportProgress(tree ReadOnlyTree) (uint64, error) {
+	progBytes, err := tree.Get([]byte("aImportProgress"))
+	if err != nil {
+		return 0, stacktrace.Propagate(err, "")
+	}
+	if len(progBytes) != 8 {
+		return 0, nil
+	}
+	return binary.BigEndian.Uint64(progBytes), nil
+}
+
+func (t *TreeStore) SetAuthoritativeImportProgress(tree *iavl.MutableTree, nextCursor uint64) error {
+	value := make([]byte, 8)
+	binary.BigEndian.PutUint64(value, nextCursor)
+
+	_, err := tree.Set([]byte("aImportProgress"), value)
+	return stacktrace.Propagate(err, "")
+}