go: store/datas/database_common: store/nbs/store.go: Fix some issues when pushing to a dolt sql-server that is itself running GC.

go/libraries/doltcore/env/actions/remotes.go

@@ -245,7 +245,6 @@ func PushToRemoteBranch[C doltdb.Context](ctx C, rsr env.RepoStateReader[C], tem
 
 	switch err {
 	case nil:
-		cli.Println()
 		return nil
 	case doltdb.ErrUpToDate, doltdb.ErrIsAhead, ErrCantFF, datas.ErrMergeNeeded, datas.ErrDirtyWorkspace, ErrShallowPushImpossible:
 		return err

go/store/datas/database_common.go

@@ -279,6 +279,11 @@ func (db *database) doSetHead(ctx context.Context, ds Dataset, addr hash.Hash, w
 		return err
 	}
 
+	if newHead == nil {
+		// This can happen on an attempt to set a head to an address which does not exist in the database.
+		return fmt.Errorf("SetHead failed: attempt to set a dataset head to an address which is not in the store")
+	}
+
 	newVal := newHead.value()
 
 	headType := newHead.TypeName()

go/store/nbs/file_table_persister.go

@@ -25,6 +25,7 @@ import (
 	"bytes"
 	"context"
 	"errors"
+	"fmt"
 	"io"
 	"io/fs"
 	"os"
@@ -133,8 +134,8 @@ func (ftp *fsTablePersister) CopyTableFile(ctx context.Context, r io.Reader, fil
 
 		defer func() {
 			cerr := temp.Close()
-			if err == nil {
-				err = cerr
+			if cerr != nil {
+				err = errors.Join(err, fmt.Errorf("error Closing temp in CopyTableFile: %w", cerr))
 			}
 		}()
 
@@ -197,9 +198,9 @@ func (ftp *fsTablePersister) persistTable(ctx context.Context, name hash.Hash, d
 		}
 
 		defer func() {
-			closeErr := temp.Close()
-			if ferr == nil {
-				ferr = closeErr
+			cerr := temp.Close()
+			if cerr != nil {
+				ferr = errors.Join(ferr, fmt.Errorf("error Closing temp in persistTable: %w", cerr))
 			}
 		}()
 
@@ -408,7 +409,7 @@ func (ftp *fsTablePersister) PruneTableFiles(ctx context.Context, keeper func()
 		if _, ok := ftp.curTmps[filepath.Clean(p)]; !ok {
 			err := file.Remove(p)
 			if err != nil && !errors.Is(err, fs.ErrNotExist) {
-				ea.add(p, err)
+				ea.add(p, fmt.Errorf("error file.Remove unfilteredTempFiles: %w", err))
 			}
 		}
 		ftp.removeMu.Unlock()
@@ -419,7 +420,7 @@ func (ftp *fsTablePersister) PruneTableFiles(ctx context.Context, keeper func()
 		if _, ok := ftp.toKeep[filepath.Clean(p)]; !ok {
 			err := file.Remove(p)
 			if err != nil && !errors.Is(err, fs.ErrNotExist) {
-				ea.add(p, err)
+				ea.add(p, fmt.Errorf("error file.Remove unfilteredTableFiles: %w", err))
 			}
 		}
 		ftp.removeMu.Unlock()

go/store/nbs/file_table_reader.go

@@ -29,6 +29,7 @@ import (
 	"os"
 	"path/filepath"
 	"strings"
+	"sync/atomic"
 	"time"
 
 	"github.com/dolthub/dolt/go/store/hash"
@@ -107,7 +108,9 @@ func newFileReaderAt(path string) (*fileReaderAt, error) {
 		// Size returns the number of bytes for regular files and is system dependent for others (Some of which can be negative).
 		return nil, fmt.Errorf("%s has invalid size: %d", path, fi.Size())
 	}
-	return &fileReaderAt{f, path, fi.Size()}, nil
+	cnt := new(int32)
+	*cnt = 1
+	return &fileReaderAt{f, path, fi.Size(), cnt}, nil
 }
 
 func nomsFileTableReader(ctx context.Context, path string, h hash.Hash, chunkCount uint32, q MemoryQuotaProvider) (cs chunkSource, err error) {
@@ -187,22 +190,31 @@ type fileReaderAt struct {
 	f    *os.File
 	path string
 	sz   int64
+	// refcnt, clone() increments and Close() decrements. The *os.File is closed when it reaches 0.
+	cnt *int32
 }
 
 func (fra *fileReaderAt) clone() (tableReaderAt, error) {
-	f, err := os.Open(fra.path)
-	if err != nil {
-		return nil, err
+	if atomic.AddInt32(fra.cnt, 1) == 1 {
+		panic("attempt to clone a closed fileReaderAt")
 	}
 	return &fileReaderAt{
-		f,
+		fra.f,
 		fra.path,
 		fra.sz,
+		fra.cnt,
 	}, nil
 }
 
 func (fra *fileReaderAt) Close() error {
-	return fra.f.Close()
+	cnt := atomic.AddInt32(fra.cnt, -1)
+	if cnt == 0 {
+		return fra.f.Close()
+	} else if cnt < 0 {
+		panic("invalid cnt on fileReaderAt")
+	} else {
+		return nil
+	}
 }
 
 func (fra *fileReaderAt) Reader(ctx context.Context) (io.ReadCloser, error) {

go/store/nbs/store.go

@@ -216,14 +216,14 @@ func (nbs *NomsBlockStore) getChunkLocations(ctx context.Context, hashes hash.Ha
 		ranges := make(map[*chunkSource]map[hash.Hash]Range)
 
 		gcb, err := fn(tables.upstream, gr, ranges, keeper)
-		if needsContinue, err := nbs.handleUnlockedRead(ctx, gcb, endRead, err); err != nil {
+		if needsContinue, err := nbs.handleUnlockedRead(ctx, gcb, false, endRead, err); err != nil {
 			return nil, err
 		} else if needsContinue {
 			continue
 		}
 
 		gcb, err = fn(tables.novel, gr, ranges, keeper)
-		if needsContinue, err := nbs.handleUnlockedRead(ctx, gcb, endRead, err); err != nil {
+		if needsContinue, err := nbs.handleUnlockedRead(ctx, gcb, true, endRead, err); err != nil {
 			return nil, err
 		} else if needsContinue {
 			continue
@@ -251,7 +251,7 @@ func (nbs *NomsBlockStore) GetChunkLocations(ctx context.Context, hashes hash.Ha
 	return res, nil
 }
 
-func (nbs *NomsBlockStore) handleUnlockedRead(ctx context.Context, gcb gcBehavior, endRead func(), err error) (bool, error) {
+func (nbs *NomsBlockStore) handleUnlockedRead(ctx context.Context, gcb gcBehavior, endReadOnSuccess bool, endRead func(), err error) (bool, error) {
 	if err != nil {
 		if endRead != nil {
 			nbs.mu.Lock()
@@ -269,7 +269,7 @@ func (nbs *NomsBlockStore) handleUnlockedRead(ctx context.Context, gcb gcBehavio
 		nbs.mu.Unlock()
 		return true, err
 	} else {
-		if endRead != nil {
+		if endRead != nil && endReadOnSuccess {
 			nbs.mu.Lock()
 			endRead()
 			nbs.mu.Unlock()
@@ -919,7 +919,7 @@ func (nbs *NomsBlockStore) Get(ctx context.Context, h hash.Hash) (chunks.Chunk,
 		nbs.mu.Unlock()
 
 		data, gcb, err := tables.get(ctx, h, keeper, nbs.stats)
-		needContinue, err := nbs.handleUnlockedRead(ctx, gcb, endRead, err)
+		needContinue, err := nbs.handleUnlockedRead(ctx, gcb, true, endRead, err)
 		if err != nil {
 			return chunks.EmptyChunk, err
 		}
@@ -1014,7 +1014,7 @@ func (nbs *NomsBlockStore) getManyWithFunc(
 			_, gcb, err := getManyFunc(ctx, tables, eg, reqs, keeper, nbs.stats)
 			return gcb, errors.Join(err, eg.Wait())
 		}()
-		needContinue, err := nbs.handleUnlockedRead(ctx, gcb, endRead, err)
+		needContinue, err := nbs.handleUnlockedRead(ctx, gcb, true, endRead, err)
 		if err != nil {
 			return err
 		}
@@ -1103,7 +1103,7 @@ func (nbs *NomsBlockStore) Has(ctx context.Context, h hash.Hash) (bool, error) {
 		nbs.mu.Unlock()
 
 		has, gcb, err := tables.has(h, keeper)
-		needsContinue, err := nbs.handleUnlockedRead(ctx, gcb, endRead, err)
+		needsContinue, err := nbs.handleUnlockedRead(ctx, gcb, true, endRead, err)
 		if err != nil {
 			return false, err
 		}
@@ -1165,7 +1165,7 @@ func (nbs *NomsBlockStore) hasManyDep(ctx context.Context, hashes hash.HashSet,
 		nbs.mu.Unlock()
 
 		remaining, gcb, err := tables.hasMany(reqs, keeper)
-		needContinue, err := nbs.handleUnlockedRead(ctx, gcb, endRead, err)
+		needContinue, err := nbs.handleUnlockedRead(ctx, gcb, true, endRead, err)
 		if err != nil {
 			return nil, err
 		}
@@ -1804,7 +1804,7 @@ func (nbs *NomsBlockStore) addTableFilesToManifest(ctx context.Context, fileIdTo
 		// checks pass.
 		sources, err := nbs.openChunkSourcesForAddTableFiles(ctx, fileIdHashToNumChunks)
 		if err != nil {
-			return err
+			return fmt.Errorf("addTableFiles, openChunkSources: %w", err)
 		}
 		// If these sources get added to the store, they will get cloned.
 		// Either way, we want to close these instances when we are done.
@@ -1826,15 +1826,15 @@ func (nbs *NomsBlockStore) addTableFilesToManifest(ctx context.Context, fileIdTo
 			err = refCheckAllSources(ctx, getAddrs, refCheck, sources.sources, nbs.stats)
 			if err != nil {
 				// There was an error checking all references.
-				return err
+				return fmt.Errorf("addTableFiles, refCheckAllSources: %w", err)
 			}
 		}
 
 		// At this point, the added files are consistent with our view of the store.
 		// We add them to the set of table files in the store.
 		_, gcGenMismatch, err := nbs.updateManifestAddFiles(ctx, fileIdHashToNumChunks, nil, &sources.gcGen, sources.sources)
 		if err != nil {
-			return err
+			return fmt.Errorf("addTableFiles, updateManifestAddFiles: %w", err)
 		} else if gcGenMismatch {
 			// A gcGenMismatch means that the store has changed out from under
 			// us as we were running these checks. We want to retry.
@@ -2233,13 +2233,13 @@ func (nbs *NomsBlockStore) swapTables(ctx context.Context, specs []tableSpec, mo
 	// replace nbs.tables.upstream with gc compacted tables
 	ts, err := nbs.tables.rebase(ctx, upstream.specs, nil, nbs.stats)
 	if err != nil {
-		return err
+		return fmt.Errorf("swapTables, rebase: %w", err)
 	}
 	oldTables := nbs.tables
 	nbs.tables, nbs.upstream = ts, upstream
 	err = oldTables.close()
 	if err != nil {
-		return err
+		return fmt.Errorf("swapTables, oldTables.close(): %w", err)
 	}
 
 	// When this is called, we are at a safepoint in the GC process.
@@ -2250,13 +2250,7 @@ func (nbs *NomsBlockStore) swapTables(ctx context.Context, specs []tableSpec, mo
 	for _, css := range oldNovel {
 		err = css.close()
 		if err != nil {
-			return err
-		}
-	}
-	if nbs.memtable != nil {
-		var thrown []string
-		for a := range nbs.memtable.chunks {
-			thrown = append(thrown, a.String())
+			return fmt.Errorf("swapTables, oldNovel css.close(): %w", err)
 		}
 	}
 	nbs.memtable = nil

go/store/nbs/table_reader.go

@@ -742,12 +742,7 @@ func (tr tableReader) currentSize() uint64 {
 }
 
 func (tr tableReader) close() error {
-	err := tr.idx.Close()
-	if err != nil {
-		tr.r.Close()
-		return err
-	}
-	return tr.r.Close()
+	return errors.Join(tr.idx.Close(), tr.r.Close())
 }
 
 func (tr tableReader) clone() (tableReader, error) {

go/store/nbs/table_set.go

@@ -338,23 +338,14 @@ func (ts tableSet) physicalLen() (uint64, error) {
 	return lenNovel + lenUp, nil
 }
 
-func (ts tableSet) close() error {
-	var firstErr error
-	setErr := func(err error) {
-		if err != nil && firstErr == nil {
-			firstErr = err
-		}
-	}
-
+func (ts tableSet) close() (err error) {
 	for _, t := range ts.novel {
-		err := t.close()
-		setErr(err)
+		err = errors.Join(err, t.close())
 	}
 	for _, t := range ts.upstream {
-		err := t.close()
-		setErr(err)
+		err = errors.Join(err, t.close())
 	}
-	return firstErr
+	return
 }
 
 // Size returns the number of tables in this tableSet.