fix(base.SaveDataset): move logbook write operation back down from lib

This fixes a few tests that were failing thanks to one-off save implmentations
that no longer received logbook updates with logbook write operations moved up
into lib. This was pretty tough to track down

Interaction between subsystems on the save path has become hard to predict. We
have code that implictly require base.SaveDataset manages updates to history
changes. It's test code, but the problem is the implicit-ness, and brittleness
of the codebase

I think th eway forward on both is to use the event bus, and focus on wiring the
bus up properly in all circumstances. If logbook updates from save events
happened via a blocking event bus subscription we could avoid this brittleness
As-is we need to plumb a *run.State argument down into base.SaveDataset just to
pass along to logbook's write.

base/dataset_prepare.go

@@ -98,7 +98,7 @@ func PrepareSaveRef(
 		}
 
 		// we have a valid previous reference & an initID, return!
-		log.Debugf("PrepareSaveRef found previous initID=%q", ref.InitID)
+		log.Debugw("PrepareSaveRef found previous initID", "initID", ref.InitID, "path", ref.Path)
 		return ref, false, nil
 	}
 

base/save.go

@@ -10,6 +10,7 @@ import (
 	"github.com/qri-io/qri/dsref"
 	"github.com/qri-io/qri/profile"
 	"github.com/qri-io/qri/repo"
+	"github.com/qri-io/qri/transform/run"
 )
 
 // SaveSwitches is an alias for the switches that control how saves happen
@@ -19,7 +20,16 @@ type SaveSwitches = dsfs.SaveSwitches
 var ErrNameTaken = fmt.Errorf("name already in use")
 
 // SaveDataset saves a version of the dataset for the given initID at the current path
-func SaveDataset(ctx context.Context, r repo.Repo, writeDest qfs.Filesystem, initID, prevPath string, changes *dataset.Dataset, sw SaveSwitches) (ds *dataset.Dataset, err error) {
+func SaveDataset(
+	ctx context.Context,
+	r repo.Repo,
+	writeDest qfs.Filesystem,
+	initID string,
+	prevPath string,
+	changes *dataset.Dataset,
+	runState *run.State,
+	sw SaveSwitches,
+) (ds *dataset.Dataset, err error) {
 	log.Debugf("SaveDataset initID=%q prevPath=%q", initID, prevPath)
 	var pro *profile.Profile
 	if pro, err = r.Profile(ctx); err != nil {
@@ -103,6 +113,10 @@ func SaveDataset(ctx context.Context, r repo.Repo, writeDest qfs.Filesystem, ini
 		return nil, err
 	}
 
+	// Write the save to logbook
+	if err = r.Logbook().WriteVersionSave(ctx, initID, ds, runState); err != nil {
+		return nil, err
+	}
 	return ds, nil
 }
 

base/test_runner_test.go

@@ -62,7 +62,7 @@ func (run *TestRunner) saveDataset(ds *dataset.Dataset, sw SaveSwitches) (dsref.
 		return dsref.Ref{}, err
 	}
 
-	ds, err := SaveDataset(run.Context, run.Repo, run.Repo.Filesystem().DefaultWriteFS(), ref.InitID, ref.Path, ds, sw)
+	ds, err := SaveDataset(run.Context, run.Repo, run.Repo.Filesystem().DefaultWriteFS(), ref.InitID, ref.Path, ds, nil, sw)
 	if err != nil {
 		return dsref.Ref{}, err
 	}

cmd/test_runner_test.go

@@ -652,7 +652,7 @@ func (run *TestRunner) AddDatasetToRefstore(t *testing.T, refStr string, ds *dat
 	// No existing commit
 	emptyHeadRef := ""
 
-	if _, err = base.SaveDataset(ctx, r, r.Filesystem().DefaultWriteFS(), initID, emptyHeadRef, ds, base.SaveSwitches{}); err != nil {
+	if _, err = base.SaveDataset(ctx, r, r.Filesystem().DefaultWriteFS(), initID, emptyHeadRef, ds, nil, base.SaveSwitches{}); err != nil {
 		t.Fatal(err)
 	}
 

lib/datasets.go

@@ -737,7 +737,7 @@ func (p *SaveParams) SetNonZeroDefaults() {
 
 // Save adds a history entry, updating a dataset
 func (m *DatasetMethods) Save(ctx context.Context, p *SaveParams) (*dataset.Dataset, error) {
-	log.Debugf("DatasetMethods.Save p=%v", p)
+	log.Debugw("DatasetMethods.Save", "params", p)
 	res := &dataset.Dataset{}
 
 	if m.inst.http != nil {
@@ -938,7 +938,7 @@ func (m *DatasetMethods) Save(ctx context.Context, p *SaveParams) (*dataset.Data
 		NewName:             p.NewName,
 		Drop:                p.Drop,
 	}
-	savedDs, err := base.SaveDataset(ctx, m.inst.repo, writeDest, ref.InitID, ref.Path, ds, switches)
+	savedDs, err := base.SaveDataset(ctx, m.inst.repo, writeDest, ref.InitID, ref.Path, ds, runState, switches)
 	if err != nil {
 		// datasets that are unchanged & have a runState record a record of no-changes
 		// to logbook
@@ -955,11 +955,6 @@ func (m *DatasetMethods) Save(ctx context.Context, p *SaveParams) (*dataset.Data
 		return nil, err
 	}
 
-	// Write the save to logbook
-	if err = m.inst.logbook.WriteVersionSave(ctx, ref.InitID, savedDs, runState); err != nil {
-		return nil, err
-	}
-
 	success = true
 
 	// TODO (b5) - this should be integrated into base.SaveDataset

lib/datasets_test.go

@@ -165,7 +165,7 @@ func TestDatasetRequestsForceSave(t *testing.T) {
 		Force: true,
 	})
 	if err != nil {
-		t.Errorf("expected empty save with flag to not error. got: %s", err.Error())
+		t.Errorf("expected empty save with force flag to not error. got: %q", err.Error())
 	}
 }
 

lib/lib.go

@@ -663,10 +663,11 @@ func NewInstanceFromConfigAndNodeAndBus(ctx context.Context, cfg *config.Config,
 		cancel: cancel,
 		doneCh: make(chan struct{}),
 
-		cfg:     cfg,
-		node:    node,
-		dscache: dc,
-		logbook: r.Logbook(),
+		cfg:       cfg,
+		node:      node,
+		dscache:   dc,
+		logbook:   r.Logbook(),
+		transform: transform.NewService(ctx),
 	}
 
 	inst.stats = stats.New(nil)

lib/lib_test.go

@@ -222,7 +222,7 @@ func saveDataset(ctx context.Context, r repo.Repo, ds *dataset.Dataset, sw base.
 	if err != nil {
 		return dsref.Ref{}, err
 	}
-	res, err := base.SaveDataset(ctx, r, r.Filesystem().DefaultWriteFS(), ref.InitID, ref.Path, ds, sw)
+	res, err := base.SaveDataset(ctx, r, r.Filesystem().DefaultWriteFS(), ref.InitID, ref.Path, ds, nil, sw)
 	if err != nil {
 		return dsref.Ref{}, err
 	}

lib/transform_test.go

@@ -18,7 +18,7 @@ func TestApplyTransform(t *testing.T) {
 		BodyPath: "testdata/cities_2/body.csv",
 	})
 	if err != nil {
-		t.Error(err)
+		t.Fatal(err)
 	}
 
 	// Apply a transformation

logbook/logbook.go

@@ -513,7 +513,7 @@ func (book *Book) WriteVersionSave(ctx context.Context, initID string, ds *datas
 		return ErrNoLogbook
 	}
 
-	log.Debugf("WriteVersionSave: %s", initID)
+	log.Debugw("WriteVersionSave", "initID", initID)
 	branchLog, err := book.branchLog(ctx, initID)
 	if err != nil {
 		return err
@@ -610,9 +610,12 @@ func (book *Book) appendTransformRun(blog *BranchLog, rs *run.State) int {
 		Ref:   rs.ID,
 		Name:  fmt.Sprintf("%d", rs.Number),
 
-		Timestamp: rs.StartTime.UnixNano(),
-		Size:      int64(rs.Duration),
-		Note:      string(rs.Status),
+		Size: int64(rs.Duration),
+		Note: string(rs.Status),
+	}
+
+	if rs.StartTime != nil {
+		op.Timestamp = rs.StartTime.UnixNano()
 	}
 
 	blog.Append(op)
@@ -871,10 +874,12 @@ func (book *Book) ResolveRef(ctx context.Context, ref *dsref.Ref) (string, error
 
 	var branchLog *BranchLog
 	if ref.Path == "" {
+		log.Debugw("finding branch log", "initID", initID)
 		branchLog, err = book.branchLog(ctx, initID)
 		if err != nil {
 			return "", err
 		}
+		log.Debugw("found branch log", "initID", initID, "size", branchLog.Size(), "latestSavePath", book.latestSavePath(branchLog.l))
 		ref.Path = book.latestSavePath(branchLog.l)
 	}
 

remote/remote_test.go

@@ -524,7 +524,7 @@ func saveDataset(ctx context.Context, r repo.Repo, peername string, ds *dataset.
 	if err != nil {
 		panic(err)
 	}
-	res, err := base.SaveDataset(ctx, r, r.Filesystem().DefaultWriteFS(), initID, headRef, ds, base.SaveSwitches{})
+	res, err := base.SaveDataset(ctx, r, r.Filesystem().DefaultWriteFS(), initID, headRef, ds, nil, base.SaveSwitches{})
 	if err != nil {
 		panic(err)
 	}