fix(logbook): remove 'stranded' log histories on new dataset creation

if a user starts the save process but abandons it midway with SIGKILL qri's currently
structured to terminate without any cleanup. This means the 'rollback' preocedure in
an errored save never fires, and the dataset name is left taken but not in the user's
list of datasets. This (admittedly hacky) solution checks for stray histories in this
exact state (name and branch logs declared, no version save) and considers them safe
to remove before creating a new replacement log.

logbook/logbook.go

@@ -284,6 +284,9 @@ func (book *Book) WriteAuthorRename(ctx context.Context, newName string) error {
 }
 
 // WriteDatasetInit initializes a new dataset name within the author's namespace
+// TODO (b5) - this should accept a username param. In the future "org" type
+// users will want to delegate dataset creation to different keys, where the org
+// username is used
 func (book *Book) WriteDatasetInit(ctx context.Context, dsName string) (string, error) {
 	if book == nil {
 		return "", ErrNoLogbook
@@ -294,8 +297,16 @@ func (book *Book) WriteDatasetInit(ctx context.Context, dsName string) (string,
 	if !dsref.IsValidName(dsName) {
 		return "", fmt.Errorf("logbook: dataset name %q invalid", dsName)
 	}
-	if _, err := book.DatasetRef(ctx, dsref.Ref{Username: book.Username(), Name: dsName}); err == nil {
-		return "", fmt.Errorf("logbook: dataset named %q already exists", dsName)
+	if dsLog, err := book.DatasetRef(ctx, dsref.Ref{Username: book.Username(), Name: dsName}); err == nil {
+		// check for "blank" logs, and remove them
+		if len(dsLog.Ops) == 1 && len(dsLog.Logs) == 1 && len(dsLog.Logs[0].Ops) == 1 {
+			log.Debugw("removing stranded reference", "ref", dsref.Ref{Username: book.Username(), Name: dsName})
+			if err := book.RemoveLog(ctx, dsref.Ref{Username: book.Username(), Name: dsName}); err != nil {
+				return "", fmt.Errorf("logbook: removing stray log: %w", err)
+			}
+		} else {
+			return "", fmt.Errorf("logbook: dataset named %q already exists", dsName)
+		}
 	}
 
 	authorLog, err := book.authorLog(ctx)

logbook/logbook_test.go

@@ -543,9 +543,7 @@ func TestDatasetLogNaming(t *testing.T) {
 	if err != nil {
 		t.Fatalf("unexpected error writing valid dataset name: %s", err)
 	}
-	if _, err = tr.Book.WriteDatasetInit(tr.Ctx, "airport_codes"); err == nil {
-		t.Error("expected initializing a name that already exists to error")
-	}
+
 	if err = tr.Book.WriteDatasetRename(tr.Ctx, firstInitID, "iata_airport_codes"); err != nil {
 		t.Errorf("unexpected error renaming dataset: %s", err)
 	}
@@ -625,6 +623,29 @@ func TestDatasetLogNaming(t *testing.T) {
 	if diff := cmp.Diff(expect, got); diff != "" {
 		t.Errorf("result mismatch (-want +got):\n%s", diff)
 	}
+
+	if _, err = tr.Book.WriteDatasetInit(tr.Ctx, "overwrite"); err != nil {
+		t.Fatalf("unexpected error writing valid dataset name: %s", err)
+	}
+	if _, err = tr.Book.WriteDatasetInit(tr.Ctx, "overwrite"); err != nil {
+		t.Fatalf("unexpected error overwrite an empty dataset history: %s", err)
+	}
+	err = tr.Book.WriteVersionSave(tr.Ctx, firstInitID, &dataset.Dataset{
+		Peername: tr.Username,
+		Name:     "atmospheric_particulates",
+		Commit: &dataset.Commit{
+			Timestamp: time.Date(2000, time.January, 1, 0, 0, 0, 0, time.UTC),
+			Title:     "initial commit",
+		},
+		Path: "HashOfVersion1",
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if _, err = tr.Book.WriteDatasetInit(tr.Ctx, "overwrite"); err != nil {
+		t.Error("expected initializing a name that exists with a history to error")
+	}
 }
 
 func TestBookPlainLogs(t *testing.T) {