tvx extract: more tipset extraction goodness.

- ability to extract a tipset range into individual vectors.
- ability to extract a tipset range and squash into a single multi-tipset vector.
- mark statediff output deterministically, so it can be extracted by tooling.
- ability to execute callbacks between tipsets in the driver.
- implement save-balances callback.

cmd/tvx/exec.go

@@ -1,33 +1,48 @@
 package main
 
 import (
+	"bufio"
 	"encoding/json"
 	"fmt"
 	"io"
 	"log"
 	"os"
+	"path/filepath"
+	"strings"
 
 	"github.com/fatih/color"
+	"github.com/filecoin-project/go-address"
+	cbornode "github.com/ipfs/go-ipld-cbor"
 	"github.com/urfave/cli/v2"
 
-	"github.com/filecoin-project/lotus/conformance"
-
 	"github.com/filecoin-project/test-vectors/schema"
+
+	"github.com/filecoin-project/lotus/chain/state"
+	"github.com/filecoin-project/lotus/chain/types"
+	"github.com/filecoin-project/lotus/conformance"
+	"github.com/filecoin-project/lotus/lib/blockstore"
 )
 
 var execFlags struct {
 	file               string
+	out                string
+	driverOpts         cli.StringSlice
 	fallbackBlockstore bool
 }
 
+const (
+	optSaveBalances = "save-balances"
+)
+
 var execCmd = &cli.Command{
 	Name:        "exec",
-	Description: "execute one or many test vectors against Lotus; supplied as a single JSON file, or a ndjson stdin stream",
-	Action:      runExecLotus,
+	Description: "execute one or many test vectors against Lotus; supplied as a single JSON file, a directory, or a ndjson stdin stream",
+	Action:      runExec,
 	Flags: []cli.Flag{
+		&repoFlag,
 		&cli.StringFlag{
 			Name:        "file",
-			Usage:       "input file; if not supplied, the vector will be read from stdin",
+			Usage:       "input file or directory; if not supplied, the vector will be read from stdin",
 			TakesFile:   true,
 			Destination: &execFlags.file,
 		},
@@ -36,10 +51,20 @@ var execCmd = &cli.Command{
 			Usage:       "sets the full node API as a fallback blockstore; use this if you're transplanting vectors and get block not found errors",
 			Destination: &execFlags.fallbackBlockstore,
 		},
+		&cli.StringFlag{
+			Name:        "out",
+			Usage:       "output directory where to save the results, only used when the input is a directory",
+			Destination: &execFlags.out,
+		},
+		&cli.StringSliceFlag{
+			Name:        "driver-opt",
+			Usage:       "comma-separated list of driver options (EXPERIMENTAL; will change), supported: 'save-balances=<dst>', 'pipeline-basefee' (unimplemented); only available in single-file mode",
+			Destination: &execFlags.driverOpts,
+		},
 	},
 }
 
-func runExecLotus(c *cli.Context) error {
+func runExec(c *cli.Context) error {
 	if execFlags.fallbackBlockstore {
 		if err := initialize(c); err != nil {
 			return fmt.Errorf("fallback blockstore was enabled, but could not resolve lotus API endpoint: %w", err)
@@ -48,30 +73,97 @@ func runExecLotus(c *cli.Context) error {
 		conformance.FallbackBlockstoreGetter = FullAPI
 	}
 
-	if file := execFlags.file; file != "" {
-		// we have a single test vector supplied as a file.
-		file, err := os.Open(file)
-		if err != nil {
-			return fmt.Errorf("failed to open test vector: %w", err)
-		}
-
-		var (
-			dec = json.NewDecoder(file)
-			tv  schema.TestVector
-		)
-
-		if err = dec.Decode(&tv); err != nil {
-			return fmt.Errorf("failed to decode test vector: %w", err)
-		}
-
-		return executeTestVector(tv)
+	path := execFlags.file
+	if path == "" {
+		return execVectorsStdin()
 	}
 
+	fi, err := os.Stat(path)
+	if err != nil {
+		return err
+	}
+
+	if fi.IsDir() {
+		// we're in directory mode; ensure the out directory exists.
+		outdir := execFlags.out
+		if outdir == "" {
+			return fmt.Errorf("no output directory provided")
+		}
+		if err := ensureDir(outdir); err != nil {
+			return err
+		}
+		return execVectorDir(path, outdir)
+	}
+
+	// process tipset vector options.
+	if err := processTipsetOpts(); err != nil {
+		return err
+	}
+
+	_, err = execVectorFile(new(conformance.LogReporter), path)
+	return err
+}
+
+func processTipsetOpts() error {
+	for _, opt := range execFlags.driverOpts.Value() {
+		switch ss := strings.Split(opt, "="); {
+		case ss[0] == optSaveBalances:
+			filename := ss[1]
+			log.Printf("saving balances after each tipset in: %s", filename)
+			balancesFile, err := os.Create(filename)
+			if err != nil {
+				return err
+			}
+			w := bufio.NewWriter(balancesFile)
+			cb := func(bs blockstore.Blockstore, params *conformance.ExecuteTipsetParams, res *conformance.ExecuteTipsetResult) {
+				cst := cbornode.NewCborStore(bs)
+				st, err := state.LoadStateTree(cst, res.PostStateRoot)
+				if err != nil {
+					return
+				}
+				_ = st.ForEach(func(addr address.Address, actor *types.Actor) error {
+					_, err := fmt.Fprintln(w, params.ExecEpoch, addr, actor.Balance)
+					return err
+				})
+				_ = w.Flush()
+			}
+			conformance.TipsetVectorOpts.OnTipsetApplied = append(conformance.TipsetVectorOpts.OnTipsetApplied, cb)
+
+		}
+
+	}
+	return nil
+}
+
+func execVectorDir(path string, outdir string) error {
+	files, err := filepath.Glob(filepath.Join(path, "*"))
+	if err != nil {
+		return fmt.Errorf("failed to glob input directory %s: %w", path, err)
+	}
+	for _, f := range files {
+		outfile := strings.TrimSuffix(filepath.Base(f), filepath.Ext(f)) + ".out"
+		outpath := filepath.Join(outdir, outfile)
+		outw, err := os.Create(outpath)
+		if err != nil {
+			return fmt.Errorf("failed to create file %s: %w", outpath, err)
+		}
+
+		log.Printf("processing vector %s; sending output to %s", f, outpath)
+		log.SetOutput(io.MultiWriter(os.Stderr, outw)) // tee the output.
+		_, _ = execVectorFile(new(conformance.LogReporter), f)
+		log.SetOutput(os.Stderr)
+		_ = outw.Close()
+	}
+	return nil
+}
+
+func execVectorsStdin() error {
+	r := new(conformance.LogReporter)
 	for dec := json.NewDecoder(os.Stdin); ; {
 		var tv schema.TestVector
 		switch err := dec.Decode(&tv); err {
 		case nil:
-			if err = executeTestVector(tv); err != nil {
+			if _, err = executeTestVector(r, tv); err != nil {
 				return err
 			}
 		case io.EOF:
@@ -84,19 +176,30 @@ func runExecLotus(c *cli.Context) error {
 	}
 }
 
-func executeTestVector(tv schema.TestVector) error {
+func execVectorFile(r conformance.Reporter, path string) (diffs []string, error error) {
+	file, err := os.Open(path)
+	if err != nil {
+		return nil, fmt.Errorf("failed to open test vector: %w", err)
+	}
+
+	var tv schema.TestVector
+	if err = json.NewDecoder(file).Decode(&tv); err != nil {
+		return nil, fmt.Errorf("failed to decode test vector: %w", err)
+	}
+	return executeTestVector(r, tv)
+}
+
+func executeTestVector(r conformance.Reporter, tv schema.TestVector) (diffs []string, err error) {
 	log.Println("executing test vector:", tv.Meta.ID)
 
 	for _, v := range tv.Pre.Variants {
-		r := new(conformance.LogReporter)
-
 		switch class, v := tv.Class, v; class {
 		case "message":
-			conformance.ExecuteMessageVector(r, &tv, &v)
+			diffs, err = conformance.ExecuteMessageVector(r, &tv, &v)
 		case "tipset":
-			conformance.ExecuteTipsetVector(r, &tv, &v)
+			diffs, err = conformance.ExecuteTipsetVector(r, &tv, &v)
 		default:
-			return fmt.Errorf("test vector class %s not supported", class)
+			return nil, fmt.Errorf("test vector class %s not supported", class)
 		}
 
 		if r.Failed() {
@@ -106,5 +209,5 @@ func executeTestVector(tv schema.TestVector) error {
 		}
 	}
 
-	return nil
+	return diffs, err
 }

cmd/tvx/extract.go

@@ -1,8 +1,14 @@
 package main
 
 import (
+	"encoding/json"
 	"fmt"
+	"io"
+	"log"
+	"os"
+	"path/filepath"
 
+	"github.com/filecoin-project/test-vectors/schema"
 	"github.com/urfave/cli/v2"
 )
 
@@ -21,6 +27,7 @@ type extractOpts struct {
 	retain             string
 	precursor          string
 	ignoreSanityChecks bool
+	squash             bool
 }
 
 var extractFlags extractOpts
@@ -62,13 +69,13 @@ var extractCmd = &cli.Command{
 		},
 		&cli.StringFlag{
 			Name:        "tsk",
-			Usage:       "tipset key to extract into a vector",
+			Usage:       "tipset key to extract into a vector, or range of tipsets in tsk1..tsk2 form",
 			Destination: &extractFlags.tsk,
 		},
 		&cli.StringFlag{
 			Name:        "out",
 			Aliases:     []string{"o"},
-			Usage:       "file to write test vector to",
+			Usage:       "file to write test vector to, or directory to write the batch to",
 			Destination: &extractFlags.file,
 		},
 		&cli.StringFlag{
@@ -93,6 +100,12 @@ var extractCmd = &cli.Command{
 			Value:       false,
 			Destination: &extractFlags.ignoreSanityChecks,
 		},
+		&cli.BoolFlag{
+			Name:        "squash",
+			Usage:       "when extracting a tipset range, squash all tipsets into a single vector",
+			Value:       false,
+			Destination: &extractFlags.squash,
+		},
 	},
 }
 
@@ -106,3 +119,43 @@ func runExtract(_ *cli.Context) error {
 		return fmt.Errorf("unsupported vector class")
 	}
 }
+
+// writeVector writes the vector into the specified file, or to stdout if
+// file is empty.
+func writeVector(vector *schema.TestVector, file string) (err error) {
+	output := io.WriteCloser(os.Stdout)
+	if file := file; file != "" {
+		dir := filepath.Dir(file)
+		if err := os.MkdirAll(dir, 0755); err != nil {
+			return fmt.Errorf("unable to create directory %s: %w", dir, err)
+		}
+		output, err = os.Create(file)
+		if err != nil {
+			return err
+		}
+		defer output.Close() //nolint:errcheck
+		defer log.Printf("wrote test vector to file: %s", file)
+	}
+
+	enc := json.NewEncoder(output)
+	enc.SetIndent("", "  ")
+	return enc.Encode(&vector)
+}
+
+// writeVectors writes each vector to a different file under the specified
+// directory.
+func writeVectors(dir string, vectors ...*schema.TestVector) error {
+	// verify the output directory exists.
+	if err := ensureDir(dir); err != nil {
+		return err
+	}
+	// write each vector to its file.
+	for _, v := range vectors {
+		id := v.Meta.ID
+		path := filepath.Join(dir, fmt.Sprintf("%s.json", id))
+		if err := writeVector(v, path); err != nil {
+			return err
+		}
+	}
+	return nil
+}

cmd/tvx/extract_message.go

@@ -4,12 +4,9 @@ import (
 	"bytes"
 	"compress/gzip"
 	"context"
-	"encoding/json"
 	"fmt"
 	"io"
 	"log"
-	"os"
-	"path/filepath"
 
 	"github.com/fatih/color"
 	"github.com/filecoin-project/go-address"
@@ -316,28 +313,7 @@ func doExtractMessage(opts extractOpts) error {
 			},
 		},
 	}
-
-	return writeVector(vector, opts.file)
-}
-
-func writeVector(vector schema.TestVector, file string) (err error) {
-	output := io.WriteCloser(os.Stdout)
-	if file := file; file != "" {
-		dir := filepath.Dir(file)
-		if err := os.MkdirAll(dir, 0755); err != nil {
-			return fmt.Errorf("unable to create directory %s: %w", dir, err)
-		}
-		output, err = os.Create(file)
-		if err != nil {
-			return err
-		}
-		defer output.Close() //nolint:errcheck
-		defer log.Printf("wrote test vector to file: %s", file)
-	}
-
-	enc := json.NewEncoder(output)
-	enc.SetIndent("", "  ")
-	return enc.Encode(&vector)
+	return writeVector(&vector, opts.file)
 }
 
 // resolveFromChain queries the chain for the provided message, using the block CID to

cmd/tvx/extract_tipset.go

@@ -6,10 +6,12 @@ import (
 	"context"
 	"fmt"
 	"log"
+	"strings"
 
 	"github.com/filecoin-project/test-vectors/schema"
 	"github.com/ipfs/go-cid"
 
+	"github.com/filecoin-project/lotus/chain/types"
 	lcli "github.com/filecoin-project/lotus/cli"
 	"github.com/filecoin-project/lotus/conformance"
 )
@@ -17,170 +19,259 @@ import (
 func doExtractTipset(opts extractOpts) error {
 	ctx := context.Background()
 
-	if opts.tsk == "" {
-		return fmt.Errorf("tipset key cannot be empty")
-	}
-
 	if opts.retain != "accessed-cids" {
 		return fmt.Errorf("tipset extraction only supports 'accessed-cids' state retention")
 	}
 
-	ts, err := lcli.ParseTipSetRef(ctx, FullAPI, opts.tsk)
-	if err != nil {
-		return fmt.Errorf("failed to fetch tipset: %w", err)
+	if opts.tsk == "" {
+		return fmt.Errorf("tipset key cannot be empty")
 	}
 
-	log.Printf("tipset block count: %d", len(ts.Blocks()))
-
-	var blocks []schema.Block
-	for _, b := range ts.Blocks() {
-		msgs, err := FullAPI.ChainGetBlockMessages(ctx, b.Cid())
+	ss := strings.Split(opts.tsk, "..")
+	switch len(ss) {
+	case 1: // extracting a single tipset.
+		ts, err := lcli.ParseTipSetRef(ctx, FullAPI, opts.tsk)
 		if err != nil {
-			return fmt.Errorf("failed to get block messages (cid: %s): %w", b.Cid(), err)
+			return fmt.Errorf("failed to fetch tipset: %w", err)
+		}
+		v, err := extractTipsets(ctx, ts)
+		if err != nil {
+			return err
+		}
+		return writeVector(v, opts.file)
+
+	case 2: // extracting a range of tipsets.
+		left, err := lcli.ParseTipSetRef(ctx, FullAPI, ss[0])
+		if err != nil {
+			return fmt.Errorf("failed to fetch tipset %s: %w", ss[0], err)
+		}
+		right, err := lcli.ParseTipSetRef(ctx, FullAPI, ss[1])
+		if err != nil {
+			return fmt.Errorf("failed to fetch tipset %s: %w", ss[1], err)
 		}
 
-		log.Printf("block %s has %d messages", b.Cid(), len(msgs.Cids))
+		// resolve the tipset range.
+		tss, err := resolveTipsetRange(ctx, left, right)
+		if err != nil {
+			return err
+		}
 
-		packed := make([]schema.Base64EncodedBytes, 0, len(msgs.Cids))
-		for _, m := range msgs.BlsMessages {
-			b, err := m.Serialize()
+		// are are squashing all tipsets into a single multi-tipset vector?
+		if opts.squash {
+			vector, err := extractTipsets(ctx, tss...)
 			if err != nil {
-				return fmt.Errorf("failed to serialize message: %w", err)
+				return err
 			}
-			packed = append(packed, b)
+			return writeVector(vector, opts.file)
 		}
-		for _, m := range msgs.SecpkMessages {
-			b, err := m.Message.Serialize()
-			if err != nil {
-				return fmt.Errorf("failed to serialize message: %w", err)
-			}
-			packed = append(packed, b)
+
+		// we are generating a single-tipset vector per tipset.
+		vectors, err := extractIndividualTipsets(ctx, tss...)
+		if err != nil {
+			return err
 		}
-		blocks = append(blocks, schema.Block{
-			MinerAddr: b.Miner,
-			WinCount:  b.ElectionProof.WinCount,
-			Messages:  packed,
-		})
+		return writeVectors(opts.file, vectors...)
+
+	default:
+		return fmt.Errorf("unrecognized tipset format")
 	}
+}
 
+func resolveTipsetRange(ctx context.Context, left *types.TipSet, right *types.TipSet) (tss []*types.TipSet, err error) {
+	// start from the right tipset and walk back the chain until the left tipset, inclusive.
+	for curr := right; curr.Key() != left.Parents(); {
+		tss = append(tss, curr)
+		curr, err = FullAPI.ChainGetTipSet(ctx, curr.Parents())
+		if err != nil {
+			return nil, fmt.Errorf("failed to get tipset %s (height: %d): %w", curr.Parents(), curr.Height()-1, err)
+		}
+	}
+	// reverse the slice.
+	for i, j := 0, len(tss)-1; i < j; i, j = i+1, j-1 {
+		tss[i], tss[j] = tss[j], tss[i]
+	}
+	return tss, nil
+}
+
+func extractIndividualTipsets(ctx context.Context, tss ...*types.TipSet) (vectors []*schema.TestVector, err error) {
+	for _, ts := range tss {
+		v, err := extractTipsets(ctx, ts)
+		if err != nil {
+			return nil, err
+		}
+		vectors = append(vectors, v)
+	}
+	return vectors, nil
+}
+
+func extractTipsets(ctx context.Context, tss ...*types.TipSet) (*schema.TestVector, error) {
 	var (
 		// create a read-through store that uses ChainGetObject to fetch unknown CIDs.
 		pst = NewProxyingStores(ctx, FullAPI)
 		g   = NewSurgeon(ctx, FullAPI, pst)
+
+		// recordingRand will record randomness so we can embed it in the test vector.
+		recordingRand = conformance.NewRecordingRand(new(conformance.LogReporter), FullAPI)
 	)
 
+	tbs, ok := pst.Blockstore.(TracingBlockstore)
+	if !ok {
+		return nil, fmt.Errorf("requested 'accessed-cids' state retention, but no tracing blockstore was present")
+	}
+
 	driver := conformance.NewDriver(ctx, schema.Selector{}, conformance.DriverOpts{
 		DisableVMFlush: true,
 	})
 
+	base := tss[0]
+	last := tss[len(tss)-1]
+
 	// this is the root of the state tree we start with.
-	root := ts.ParentState()
+	root := base.ParentState()
 	log.Printf("base state tree root CID: %s", root)
 
-	basefee := ts.Blocks()[0].ParentBaseFee
-	log.Printf("basefee: %s", basefee)
-
-	tipset := schema.Tipset{
-		BaseFee: *basefee.Int,
-		Blocks:  blocks,
+	codename := GetProtocolCodename(base.Height())
+	nv, err := FullAPI.StateNetworkVersion(ctx, base.Key())
+	if err != nil {
+		return nil, err
 	}
 
-	// recordingRand will record randomness so we can embed it in the test vector.
-	recordingRand := conformance.NewRecordingRand(new(conformance.LogReporter), FullAPI)
+	version, err := FullAPI.Version(ctx)
+	if err != nil {
+		return nil, err
+	}
 
-	log.Printf("using state retention strategy: %s", extractFlags.retain)
+	ntwkName, err := FullAPI.StateNetworkName(ctx)
+	if err != nil {
+		return nil, err
+	}
 
-	tbs, ok := pst.Blockstore.(TracingBlockstore)
-	if !ok {
-		return fmt.Errorf("requested 'accessed-cids' state retention, but no tracing blockstore was present")
+	vector := schema.TestVector{
+		Class: schema.ClassTipset,
+		Meta: &schema.Metadata{
+			ID: fmt.Sprintf("@%d..@%d", base.Height(), last.Height()),
+			Gen: []schema.GenerationData{
+				{Source: fmt.Sprintf("network:%s", ntwkName)},
+				{Source: "github.com/filecoin-project/lotus", Version: version.String()}},
+			// will be completed by extra tipset stamps.
+		},
+		Selector: schema.Selector{
+			schema.SelectorMinProtocolVersion: codename,
+		},
+		Pre: &schema.Preconditions{
+			Variants: []schema.Variant{
+				{ID: codename, Epoch: int64(base.Height()), NetworkVersion: uint(nv)},
+			},
+			StateTree: &schema.StateTree{
+				RootCID: base.ParentState(),
+			},
+		},
+		Post: &schema.Postconditions{
+			StateTree: new(schema.StateTree),
+		},
 	}
 
 	tbs.StartTracing()
 
-	params := conformance.ExecuteTipsetParams{
-		Preroot:     ts.ParentState(),
-		ParentEpoch: ts.Height() - 1,
-		Tipset:      &tipset,
-		ExecEpoch:   ts.Height(),
-		Rand:        recordingRand,
-	}
-	result, err := driver.ExecuteTipset(pst.Blockstore, pst.Datastore, params)
-	if err != nil {
-		return fmt.Errorf("failed to execute tipset: %w", err)
+	roots := []cid.Cid{base.ParentState()}
+	for i, ts := range tss {
+		log.Printf("tipset %s block count: %d", ts.Key(), len(ts.Blocks()))
+
+		var blocks []schema.Block
+		for _, b := range ts.Blocks() {
+			msgs, err := FullAPI.ChainGetBlockMessages(ctx, b.Cid())
+			if err != nil {
+				return nil, fmt.Errorf("failed to get block messages (cid: %s): %w", b.Cid(), err)
+			}
+
+			log.Printf("block %s has %d messages", b.Cid(), len(msgs.Cids))
+
+			packed := make([]schema.Base64EncodedBytes, 0, len(msgs.Cids))
+			for _, m := range msgs.BlsMessages {
+				b, err := m.Serialize()
+				if err != nil {
+					return nil, fmt.Errorf("failed to serialize message: %w", err)
+				}
+				packed = append(packed, b)
+			}
+			for _, m := range msgs.SecpkMessages {
+				b, err := m.Message.Serialize()
+				if err != nil {
+					return nil, fmt.Errorf("failed to serialize message: %w", err)
+				}
+				packed = append(packed, b)
+			}
+			blocks = append(blocks, schema.Block{
+				MinerAddr: b.Miner,
+				WinCount:  b.ElectionProof.WinCount,
+				Messages:  packed,
+			})
+		}
+
+		basefee := base.Blocks()[0].ParentBaseFee
+		log.Printf("tipset basefee: %s", basefee)
+
+		tipset := schema.Tipset{
+			BaseFee:     *basefee.Int,
+			Blocks:      blocks,
+			EpochOffset: int64(i),
+		}
+
+		params := conformance.ExecuteTipsetParams{
+			Preroot:     roots[len(roots)-1],
+			ParentEpoch: ts.Height() - 1,
+			Tipset:      &tipset,
+			ExecEpoch:   ts.Height(),
+			Rand:        recordingRand,
+		}
+
+		result, err := driver.ExecuteTipset(pst.Blockstore, pst.Datastore, params)
+		if err != nil {
+			return nil, fmt.Errorf("failed to execute tipset: %w", err)
+		}
+
+		roots = append(roots, result.PostStateRoot)
+
+		// update the vector.
+		vector.ApplyTipsets = append(vector.ApplyTipsets, tipset)
+		vector.Post.ReceiptsRoots = append(vector.Post.ReceiptsRoots, result.ReceiptsRoot)
+
+		for _, res := range result.AppliedResults {
+			vector.Post.Receipts = append(vector.Post.Receipts, &schema.Receipt{
+				ExitCode:    int64(res.ExitCode),
+				ReturnValue: res.Return,
+				GasUsed:     res.GasUsed,
+			})
+		}
+
+		vector.Meta.Gen = append(vector.Meta.Gen, schema.GenerationData{
+			Source: "tipset:" + ts.Key().String(),
+		})
 	}
 
 	accessed := tbs.FinishTracing()
 
+	//
+	// ComputeBaseFee(ctx, baseTs)
+
 	// write a CAR with the accessed state into a buffer.
 	var (
 		out = new(bytes.Buffer)
 		gw  = gzip.NewWriter(out)
 	)
-	if err := g.WriteCARIncluding(gw, accessed, ts.ParentState(), result.PostStateRoot); err != nil {
-		return err
+	if err := g.WriteCARIncluding(gw, accessed, roots...); err != nil {
+		return nil, err
 	}
 	if err = gw.Flush(); err != nil {
-		return err
+		return nil, err
 	}
 	if err = gw.Close(); err != nil {
-		return err
+		return nil, err
 	}
 
-	codename := GetProtocolCodename(ts.Height())
-	nv, err := FullAPI.StateNetworkVersion(ctx, ts.Key())
-	if err != nil {
-		return err
-	}
+	vector.Randomness = recordingRand.Recorded()
+	vector.Post.StateTree.RootCID = roots[len(roots)-1]
+	vector.CAR = out.Bytes()
 
-	version, err := FullAPI.Version(ctx)
-	if err != nil {
-		return err
-	}
-
-	ntwkName, err := FullAPI.StateNetworkName(ctx)
-	if err != nil {
-		return err
-	}
-
-	vector := schema.TestVector{
-		Class: schema.ClassTipset,
-		Meta: &schema.Metadata{
-			ID: opts.id,
-			Gen: []schema.GenerationData{
-				{Source: fmt.Sprintf("network:%s", ntwkName)},
-				{Source: fmt.Sprintf("tipset:%s", ts.Key())},
-				{Source: "github.com/filecoin-project/lotus", Version: version.String()}},
-		},
-		Selector: schema.Selector{
-			schema.SelectorMinProtocolVersion: codename,
-		},
-		Randomness: recordingRand.Recorded(),
-		CAR:        out.Bytes(),
-		Pre: &schema.Preconditions{
-			Variants: []schema.Variant{
-				{ID: codename, Epoch: int64(ts.Height()), NetworkVersion: uint(nv)},
-			},
-			BaseFee: basefee.Int,
-			StateTree: &schema.StateTree{
-				RootCID: ts.ParentState(),
-			},
-		},
-		ApplyTipsets: []schema.Tipset{tipset},
-		Post: &schema.Postconditions{
-			StateTree: &schema.StateTree{
-				RootCID: result.PostStateRoot,
-			},
-			ReceiptsRoots: []cid.Cid{result.ReceiptsRoot},
-		},
-	}
-
-	for _, res := range result.AppliedResults {
-		vector.Post.Receipts = append(vector.Post.Receipts, &schema.Receipt{
-			ExitCode:    int64(res.ExitCode),
-			ReturnValue: res.Return,
-			GasUsed:     res.GasUsed,
-		})
-	}
-
-	return writeVector(vector, opts.file)
+	return &vector, nil
 }

cmd/tvx/main.go

@@ -113,3 +113,19 @@ func destroy(_ *cli.Context) error {
 	}
 	return nil
 }
+
+func ensureDir(path string) error {
+	switch fi, err := os.Stat(path); {
+	case os.IsNotExist(err):
+		if err := os.MkdirAll(path, 0755); err != nil {
+			return fmt.Errorf("failed to create directory %s: %w", path, err)
+		}
+	case err == nil:
+		if !fi.IsDir() {
+			return fmt.Errorf("path %s is not a directory: %w", path, err)
+		}
+	default:
+		return fmt.Errorf("failed to stat directory %s: %w", path, err)
+	}
+	return nil
+}

cmd/tvx/simulate.go

@@ -202,7 +202,7 @@ func runSimulateCmd(_ *cli.Context) error {
 		},
 	}
 
-	if err := writeVector(vector, simulateFlags.out); err != nil {
+	if err := writeVector(&vector, simulateFlags.out); err != nil {
 		return fmt.Errorf("failed to write vector: %w", err)
 	}
 

cmd/tvx/stores.go

@@ -149,3 +149,14 @@ func (pb *proxyingBlockstore) Put(block blocks.Block) error {
 	pb.lk.Unlock()
 	return pb.Blockstore.Put(block)
 }
+
+func (pb *proxyingBlockstore) PutMany(blocks []blocks.Block) error {
+	pb.lk.Lock()
+	if pb.tracing {
+		for _, b := range blocks {
+			pb.traced[b.Cid()] = struct{}{}
+		}
+	}
+	pb.lk.Unlock()
+	return pb.Blockstore.PutMany(blocks)
+}

conformance/corpus_test.go

@@ -11,7 +11,7 @@ import (
 	"github.com/filecoin-project/test-vectors/schema"
 )
 
-var invokees = map[schema.Class]func(Reporter, *schema.TestVector, *schema.Variant){
+var invokees = map[schema.Class]func(Reporter, *schema.TestVector, *schema.Variant) ([]string, error){
 	schema.ClassMessage: ExecuteMessageVector,
 	schema.ClassTipset:  ExecuteTipsetVector,
 }
@@ -133,7 +133,7 @@ func TestConformance(t *testing.T) {
 			for _, variant := range vector.Pre.Variants {
 				variant := variant
 				t.Run(variant.ID, func(t *testing.T) {
-					invokee(t, &vector, &variant)
+					_, _ = invokee(t, &vector, &variant) //nolint:errcheck
 				})
 			}
 		})

conformance/driver.go

@@ -71,6 +71,9 @@ type ExecuteTipsetResult struct {
 	AppliedMessages []*types.Message
 	// AppliedResults stores the results of AppliedMessages, in the same order.
 	AppliedResults []*vm.ApplyRet
+
+	// PostBaseFee returns the basefee after applying this tipset.
+	PostBaseFee abi.TokenAmount
 }
 
 type ExecuteTipsetParams struct {

conformance/runner.go

@@ -14,6 +14,7 @@ import (
 	"github.com/fatih/color"
 	"github.com/filecoin-project/go-state-types/abi"
 	"github.com/filecoin-project/go-state-types/exitcode"
+	"github.com/hashicorp/go-multierror"
 	blocks "github.com/ipfs/go-block-format"
 	"github.com/ipfs/go-blockservice"
 	"github.com/ipfs/go-cid"
@@ -38,8 +39,19 @@ var FallbackBlockstoreGetter interface {
 	ChainReadObj(context.Context, cid.Cid) ([]byte, error)
 }
 
+var TipsetVectorOpts struct {
+	// PipelineBaseFee pipelines the basefee in multi-tipset vectors from one
+	// tipset to another. Basefees in the vector are ignored, except for that of
+	// the first tipset. UNUSED.
+	PipelineBaseFee bool
+
+	// OnTipsetApplied contains callback functions called after a tipset has been
+	// applied.
+	OnTipsetApplied []func(bs blockstore.Blockstore, params *ExecuteTipsetParams, res *ExecuteTipsetResult)
+}
+
 // ExecuteMessageVector executes a message-class test vector.
-func ExecuteMessageVector(r Reporter, vector *schema.TestVector, variant *schema.Variant) {
+func ExecuteMessageVector(r Reporter, vector *schema.TestVector, variant *schema.Variant) (diffs []string, err error) {
 	var (
 		ctx       = context.Background()
 		baseEpoch = variant.Epoch
@@ -88,14 +100,16 @@ func ExecuteMessageVector(r Reporter, vector *schema.TestVector, variant *schema
 	// Once all messages are applied, assert that the final state root matches
 	// the expected postcondition root.
 	if expected, actual := vector.Post.StateTree.RootCID, root; expected != actual {
-		r.Errorf("wrong post root cid; expected %v, but got %v", expected, actual)
-		dumpThreeWayStateDiff(r, vector, bs, root)
-		r.FailNow()
+		ierr := fmt.Errorf("wrong post root cid; expected %v, but got %v", expected, actual)
+		r.Errorf(ierr.Error())
+		err = multierror.Append(err, ierr)
+		diffs = dumpThreeWayStateDiff(r, vector, bs, root)
 	}
+	return diffs, err
 }
 
 // ExecuteTipsetVector executes a tipset-class test vector.
-func ExecuteTipsetVector(r Reporter, vector *schema.TestVector, variant *schema.Variant) {
+func ExecuteTipsetVector(r Reporter, vector *schema.TestVector, variant *schema.Variant) (diffs []string, err error) {
 	var (
 		ctx       = context.Background()
 		baseEpoch = abi.ChainEpoch(variant.Epoch)
@@ -107,6 +121,7 @@ func ExecuteTipsetVector(r Reporter, vector *schema.TestVector, variant *schema.
 	bs, err := LoadBlockstore(vector.CAR)
 	if err != nil {
 		r.Fatalf("failed to load the vector CAR: %w", err)
+		return nil, err
 	}
 
 	// Create a new Driver.
@@ -118,15 +133,22 @@ func ExecuteTipsetVector(r Reporter, vector *schema.TestVector, variant *schema.
 	for i, ts := range vector.ApplyTipsets {
 		ts := ts // capture
 		execEpoch := baseEpoch + abi.ChainEpoch(ts.EpochOffset)
-		ret, err := driver.ExecuteTipset(bs, tmpds, ExecuteTipsetParams{
+		params := ExecuteTipsetParams{
 			Preroot:     root,
 			ParentEpoch: prevEpoch,
 			Tipset:      &ts,
 			ExecEpoch:   execEpoch,
 			Rand:        NewReplayingRand(r, vector.Randomness),
-		})
+		}
+		ret, err := driver.ExecuteTipset(bs, tmpds, params)
 		if err != nil {
 			r.Fatalf("failed to apply tipset %d: %s", i, err)
+			return nil, err
+		}
+
+		// invoke callbacks.
+		for _, cb := range TipsetVectorOpts.OnTipsetApplied {
+			cb(bs, &params, ret)
 		}
 
 		for j, v := range ret.AppliedResults {
@@ -136,7 +158,9 @@ func ExecuteTipsetVector(r Reporter, vector *schema.TestVector, variant *schema.
 
 		// Compare the receipts root.
 		if expected, actual := vector.Post.ReceiptsRoots[i], ret.ReceiptsRoot; expected != actual {
-			r.Errorf("post receipts root doesn't match; expected: %s, was: %s", expected, actual)
+			ierr := fmt.Errorf("post receipts root doesn't match; expected: %s, was: %s", expected, actual)
+			r.Errorf(ierr.Error())
+			err = multierror.Append(err, ierr)
 		}
 
 		prevEpoch = execEpoch
@@ -146,10 +170,12 @@ func ExecuteTipsetVector(r Reporter, vector *schema.TestVector, variant *schema.
 	// Once all messages are applied, assert that the final state root matches
 	// the expected postcondition root.
 	if expected, actual := vector.Post.StateTree.RootCID, root; expected != actual {
-		r.Errorf("wrong post root cid; expected %v, but got %v", expected, actual)
-		dumpThreeWayStateDiff(r, vector, bs, root)
-		r.FailNow()
+		ierr := fmt.Errorf("wrong post root cid; expected %v, but got %v", expected, actual)
+		r.Errorf(ierr.Error())
+		err = multierror.Append(err, ierr)
+		diffs = dumpThreeWayStateDiff(r, vector, bs, root)
 	}
+	return diffs, err
 }
 
 // AssertMsgResult compares a message result. It takes the expected receipt
@@ -169,7 +195,7 @@ func AssertMsgResult(r Reporter, expected *schema.Receipt, actual *vm.ApplyRet,
 	}
 }
 
-func dumpThreeWayStateDiff(r Reporter, vector *schema.TestVector, bs blockstore.Blockstore, actual cid.Cid) {
+func dumpThreeWayStateDiff(r Reporter, vector *schema.TestVector, bs blockstore.Blockstore, actual cid.Cid) []string {
 	// check if statediff exists; if not, skip.
 	if err := exec.Command("statediff", "--help").Run(); err != nil {
 		r.Log("could not dump 3-way state tree diff upon test failure: statediff command not found")
@@ -178,7 +204,7 @@ func dumpThreeWayStateDiff(r Reporter, vector *schema.TestVector, bs blockstore.
 		r.Log("$ cd statediff")
 		r.Log("$ go generate ./...")
 		r.Log("$ go install ./cmd/statediff")
-		return
+		return nil
 	}
 
 	tmpCar, err := writeStateToTempCAR(bs,
@@ -188,6 +214,7 @@ func dumpThreeWayStateDiff(r Reporter, vector *schema.TestVector, bs blockstore.
 	)
 	if err != nil {
 		r.Fatalf("failed to write temporary state CAR: %s", err)
+		return nil
 	}
 	defer os.RemoveAll(tmpCar) //nolint:errcheck
 
@@ -202,28 +229,43 @@ func dumpThreeWayStateDiff(r Reporter, vector *schema.TestVector, bs blockstore.
 		d3 = color.New(color.FgGreen, color.Bold).Sprint("[Δ3]")
 	)
 
-	printDiff := func(left, right cid.Cid) {
+	diff := func(left, right cid.Cid) string {
 		cmd := exec.Command("statediff", "car", "--file", tmpCar, left.String(), right.String())
 		b, err := cmd.CombinedOutput()
 		if err != nil {
 			r.Fatalf("statediff failed: %s", err)
 		}
-		r.Log(string(b))
+		return string(b)
 	}
 
 	bold := color.New(color.Bold).SprintfFunc()
 
+	r.Log(bold("-----BEGIN STATEDIFF-----"))
+
 	// run state diffs.
 	r.Log(bold("=== dumping 3-way diffs between %s, %s, %s ===", a, b, c))
 
 	r.Log(bold("--- %s left: %s; right: %s ---", d1, a, b))
-	printDiff(vector.Post.StateTree.RootCID, actual)
+	diffA := diff(vector.Post.StateTree.RootCID, actual)
+	r.Log(bold("----------BEGIN STATEDIFF A----------"))
+	r.Log(diffA)
+	r.Log(bold("----------END STATEDIFF A----------"))
 
 	r.Log(bold("--- %s left: %s; right: %s ---", d2, c, b))
-	printDiff(vector.Pre.StateTree.RootCID, actual)
+	diffB := diff(vector.Pre.StateTree.RootCID, actual)
+	r.Log(bold("----------BEGIN STATEDIFF B----------"))
+	r.Log(diffB)
+	r.Log(bold("----------END STATEDIFF B----------"))
 
 	r.Log(bold("--- %s left: %s; right: %s ---", d3, c, a))
-	printDiff(vector.Pre.StateTree.RootCID, vector.Post.StateTree.RootCID)
+	diffC := diff(vector.Pre.StateTree.RootCID, vector.Post.StateTree.RootCID)
+	r.Log(bold("----------BEGIN STATEDIFF C----------"))
+	r.Log(diffC)
+	r.Log(bold("----------END STATEDIFF C----------"))
+
+	r.Log(bold("-----END STATEDIFF-----"))
+
+	return []string{diffA, diffB, diffC}
 }
 
 // writeStateToTempCAR writes the provided roots to a temporary CAR that'll be