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/cmd/dolt/dolt.go

@@ -61,6 +61,7 @@ import (
 	"github.com/dolthub/dolt/go/libraries/events"
 	"github.com/dolthub/dolt/go/libraries/utils/argparser"
 	"github.com/dolthub/dolt/go/libraries/utils/config"
+	"github.com/dolthub/dolt/go/libraries/utils/dynassert"
 	"github.com/dolthub/dolt/go/libraries/utils/filesys"
 	"github.com/dolthub/dolt/go/store/nbs"
 	"github.com/dolthub/dolt/go/store/util/tempfiles"
@@ -191,6 +192,7 @@ const traceProf = "trace"
 const featureVersionFlag = "--feature-version"
 
 func main() {
+	dynassert.InitDyanmicAsserts()
 	os.Exit(runMain())
 }
 

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/libraries/utils/dynassert/dynassert.go

@@ -0,0 +1,61 @@
+// Copyright 2025 Dolthub, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package dynassert
+
+import (
+	"fmt"
+	"os"
+)
+
+// dynamic asserts are enabled by default so that they run during unit
+// tests, for example.
+var enabled bool = true
+
+// To be used in the top-level |main|, this disables dynamic asserts
+// unless a specific environment variable is set.
+func InitDyanmicAsserts() {
+	if os.Getenv("DOLT_ENABLE_DYNAMIC_ASSERTS") == "" {
+		enabled = false
+	}
+}
+
+// Dynamically enabled assertions. These are software integrity sanity
+// checks that are enabled for tests, both unit and integration, and
+// can be enabled anytime when we are running in a controlled
+// environment where we want to fail hard if they are violated.
+// Typically these are not enabled when `dolt` is running for its
+// users. Code making use fo dynasserts should recover gracefully even
+// when they fail.
+
+// If dynasserts are enabled and cond is false, panics with the
+// formatted string Sprintf(msg, args...).  Otherwise returns |cond|.
+//
+// A suggested usage might be something like:
+//
+//	if dynassert.Assert(atomic.AddInt32(refcnt, 1) <= 1, "invalid ref count; incremented from <= 0") {
+//	    // Restore, since we are not taking the reference...
+//	    atomic.AddInt32(refcnt, -1)
+//	    return NewObjectInstead(...)
+//	}
+//
+// return view_of_this_object_with_valid_ref
+func Assert(cond bool, msg string, args ...any) bool {
+	if enabled {
+		if !cond {
+			panic(fmt.Sprintf(msg, args...))
+		}
+	}
+	return cond
+}

go/libraries/utils/dynassert/dynassert_test.go

@@ -0,0 +1,46 @@
+// Copyright 2025 Dolthub, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package dynassert
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestAssertsAreEnabled(t *testing.T) {
+	assert.True(t, enabled)
+	Assert(true, "does not panic")
+	assert.Panics(t, func() {
+		Assert(false, "does panic")
+	})
+}
+
+func TestInitDynamicAsserts(t *testing.T) {
+	t.Run("WithoutEnvVarSet", func(t *testing.T) {
+		enabled = true
+		t.Setenv("DOLT_ENABLE_DYNAMIC_ASSERTS", "")
+		InitDyanmicAsserts()
+		assert.False(t, enabled)
+		Assert(true, "does not panic")
+		Assert(false, "does not panic")
+	})
+	t.Run("WithEnvVarSet", func(t *testing.T) {
+		enabled = true
+		t.Setenv("DOLT_ENABLE_DYNAMIC_ASSERTS", "true")
+		InitDyanmicAsserts()
+		assert.True(t, enabled)
+	})
+}

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,8 +29,10 @@ import (
 	"os"
 	"path/filepath"
 	"strings"
+	"sync/atomic"
 	"time"
 
+	"github.com/dolthub/dolt/go/libraries/utils/dynassert"
 	"github.com/dolthub/dolt/go/store/hash"
 )
 
@@ -107,7 +109,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 +191,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 dynassert.Assert(atomic.AddInt32(fra.cnt, 1) > 1, "attempt to clone a closed fileReaderAt") {
+		// Restore previous refcnt, even know we're in a weird state...
+		atomic.AddInt32(fra.cnt, -1)
+		return newFileReaderAt(fra.path)
 	}
 	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)
+	dynassert.Assert(cnt >= 0, "invalid cnt on fileReaderAt")
+	if cnt == 0 {
+		return fra.f.Close()
+	}
+	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) {