Merge pull request #8612 from dolthub/aaron/gc-gen-takes-journal-meta…

…data

go/store/nbs: Fixing GCGen to be more correct.

go/store/chunks/chunk_store.go

@@ -180,6 +180,13 @@ type GCFinalizer interface {
 	SwapChunksInStore(ctx context.Context) error
 }
 
+type GCMode int
+
+const (
+	GCMode_Default GCMode = iota
+	GCMode_Full
+)
+
 // ChunkStoreGarbageCollector is a ChunkStore that supports garbage collection.
 type ChunkStoreGarbageCollector interface {
 	ChunkStore
@@ -213,7 +220,7 @@ type ChunkStoreGarbageCollector interface {
 	// This behavior is a little different for ValueStore.GC()'s
 	// interactions with generational stores. See ValueStore and
 	// NomsBlockStore/GenerationalNBS for details.
-	MarkAndSweepChunks(ctx context.Context, hashes <-chan []hash.Hash, dest ChunkStore) (GCFinalizer, error)
+	MarkAndSweepChunks(ctx context.Context, hashes <-chan []hash.Hash, dest ChunkStore, mode GCMode) (GCFinalizer, error)
 
 	// Count returns the number of chunks in the store.
 	Count() (uint32, error)

go/store/chunks/memory_store.go

@@ -360,7 +360,7 @@ func (mgcf msvGcFinalizer) SwapChunksInStore(ctx context.Context) error {
 	return nil
 }
 
-func (ms *MemoryStoreView) MarkAndSweepChunks(ctx context.Context, hashes <-chan []hash.Hash, dest ChunkStore) (GCFinalizer, error) {
+func (ms *MemoryStoreView) MarkAndSweepChunks(ctx context.Context, hashes <-chan []hash.Hash, dest ChunkStore, mode GCMode) (GCFinalizer, error) {
 	if dest != ms {
 		panic("unsupported")
 	}

go/store/chunks/test_utils.go

@@ -91,12 +91,12 @@ func (s *TestStoreView) EndGC() {
 	collector.EndGC()
 }
 
-func (s *TestStoreView) MarkAndSweepChunks(ctx context.Context, hashes <-chan []hash.Hash, dest ChunkStore) (GCFinalizer, error) {
+func (s *TestStoreView) MarkAndSweepChunks(ctx context.Context, hashes <-chan []hash.Hash, dest ChunkStore, mode GCMode) (GCFinalizer, error) {
 	collector, ok := s.ChunkStore.(ChunkStoreGarbageCollector)
 	if !ok || dest != s {
 		return nil, ErrUnsupportedOperation
 	}
-	return collector.MarkAndSweepChunks(ctx, hashes, collector)
+	return collector.MarkAndSweepChunks(ctx, hashes, collector, mode)
 }
 
 func (s *TestStoreView) Count() (uint32, error) {

go/store/nbs/archive_build.go

@@ -105,7 +105,7 @@ func UnArchive(ctx context.Context, cs chunks.ChunkStore, smd StorageMetadata, p
 					newSpecs = append(newSpecs, spec)
 				}
 			}
-			err = gs.oldGen.swapTables(ctx, newSpecs)
+			err = gs.oldGen.swapTables(ctx, newSpecs, chunks.GCMode_Default)
 			if err != nil {
 				return err
 			}
@@ -175,7 +175,7 @@ func BuildArchive(ctx context.Context, cs chunks.ChunkStore, dagGroups *ChunkRel
 				newSpecs = append(newSpecs, spec)
 			}
 		}
-		err = gs.oldGen.swapTables(ctx, newSpecs)
+		err = gs.oldGen.swapTables(ctx, newSpecs, chunks.GCMode_Default)
 		if err != nil {
 			return err
 		}

go/store/nbs/conjoiner.go

@@ -130,7 +130,7 @@ func conjoin(ctx context.Context, s conjoinStrategy, upstream manifestContents,
 		newContents := manifestContents{
 			nbfVers:  upstream.nbfVers,
 			root:     upstream.root,
-			lock:     generateLockHash(upstream.root, specs, appendixSpecs),
+			lock:     generateLockHash(upstream.root, specs, appendixSpecs, nil),
 			gcGen:    upstream.gcGen,
 			specs:    specs,
 			appendix: appendixSpecs,

go/store/nbs/generational_chunk_store.go

@@ -503,8 +503,8 @@ func (gcs *GenerationalNBS) EndGC() {
 	gcs.newGen.EndGC()
 }
 
-func (gcs *GenerationalNBS) MarkAndSweepChunks(ctx context.Context, hashes <-chan []hash.Hash, dest chunks.ChunkStore) (chunks.GCFinalizer, error) {
-	return markAndSweepChunks(ctx, hashes, gcs.newGen, gcs, dest)
+func (gcs *GenerationalNBS) MarkAndSweepChunks(ctx context.Context, hashes <-chan []hash.Hash, dest chunks.ChunkStore, mode chunks.GCMode) (chunks.GCFinalizer, error) {
+	return markAndSweepChunks(ctx, hashes, gcs.newGen, gcs, dest, mode)
 }
 
 func (gcs *GenerationalNBS) IterateAllChunks(ctx context.Context, cb func(chunk chunks.Chunk)) error {

go/store/nbs/journal.go

@@ -206,7 +206,7 @@ func trueUpBackingManifest(ctx context.Context, root hash.Hash, backing *journal
 	}
 
 	prev := mc.lock
-	next := generateLockHash(mc.root, mc.specs, mc.appendix)
+	next := generateLockHash(mc.root, mc.specs, mc.appendix, nil)
 	mc.lock = next
 
 	mc, err = backing.Update(ctx, prev, mc, &Stats{}, nil)

go/store/nbs/manifest.go

@@ -501,7 +501,7 @@ func formatSpecs(specs []tableSpec, tableInfo []string) {
 // persisted manifest against the lock hash it saw last time it loaded the
 // contents of a manifest. If they do not match, the client must not update
 // the persisted manifest.
-func generateLockHash(root hash.Hash, specs []tableSpec, appendix []tableSpec) hash.Hash {
+func generateLockHash(root hash.Hash, specs []tableSpec, appendix []tableSpec, extra []byte) hash.Hash {
 	blockHash := sha512.New()
 	blockHash.Write(root[:])
 	for _, spec := range appendix {
@@ -511,6 +511,11 @@ func generateLockHash(root hash.Hash, specs []tableSpec, appendix []tableSpec) h
 	for _, spec := range specs {
 		blockHash.Write(spec.name[:])
 	}
+	if len(extra) > 0 {
+		blockHash.Write([]byte{0})
+		blockHash.Write(extra)
+	}
+	blockHash.Write([]byte{0})
 	var h []byte
 	h = blockHash.Sum(h) // Appends hash to h
 	return hash.New(h[:hash.ByteLen])

go/store/nbs/nbs_metrics_wrapper.go

@@ -79,8 +79,8 @@ func (nbsMW *NBSMetricWrapper) EndGC() {
 	nbsMW.nbs.EndGC()
 }
 
-func (nbsMW *NBSMetricWrapper) MarkAndSweepChunks(ctx context.Context, hashes <-chan []hash.Hash, dest chunks.ChunkStore) (chunks.GCFinalizer, error) {
-	return nbsMW.nbs.MarkAndSweepChunks(ctx, hashes, dest)
+func (nbsMW *NBSMetricWrapper) MarkAndSweepChunks(ctx context.Context, hashes <-chan []hash.Hash, dest chunks.ChunkStore, mode chunks.GCMode) (chunks.GCFinalizer, error) {
+	return nbsMW.nbs.MarkAndSweepChunks(ctx, hashes, dest, mode)
 }
 
 func (nbsMW *NBSMetricWrapper) Count() (uint32, error) {

go/store/nbs/store.go

@@ -35,6 +35,7 @@ import (
 	"cloud.google.com/go/storage"
 	"github.com/aws/aws-sdk-go/service/s3/s3iface"
 	"github.com/dustin/go-humanize"
+	"github.com/fatih/color"
 	lru "github.com/hashicorp/golang-lru/v2"
 	"github.com/oracle/oci-go-sdk/v65/common"
 	"github.com/oracle/oci-go-sdk/v65/objectstorage"
@@ -266,7 +267,7 @@ func (nbs *NomsBlockStore) UpdateManifest(ctx context.Context, updates map[hash.
 			return contents, nil
 		}
 
-		contents.lock = generateLockHash(contents.root, contents.specs, contents.appendix)
+		contents.lock = generateLockHash(contents.root, contents.specs, contents.appendix, nil)
 
 		// ensure we don't drop existing appendices
 		if contents.appendix != nil && len(contents.appendix) > 0 {
@@ -423,7 +424,7 @@ func fromManifestAppendixOptionNewContents(upstream manifestContents, appendixSp
 		newAppendixSpecs := append([]tableSpec{}, upstreamAppendixSpecs...)
 		contents.appendix = append(newAppendixSpecs, appendixSpecs...)
 
-		contents.lock = generateLockHash(contents.root, contents.specs, contents.appendix)
+		contents.lock = generateLockHash(contents.root, contents.specs, contents.appendix, nil)
 		return contents, nil
 	case ManifestAppendixOption_Set:
 		if len(appendixSpecs) < 1 {
@@ -438,7 +439,7 @@ func fromManifestAppendixOptionNewContents(upstream manifestContents, appendixSp
 		// append new appendix specs to contents.appendix
 		contents.appendix = append([]tableSpec{}, appendixSpecs...)
 
-		contents.lock = generateLockHash(contents.root, contents.specs, contents.appendix)
+		contents.lock = generateLockHash(contents.root, contents.specs, contents.appendix, nil)
 		return contents, nil
 	default:
 		return manifestContents{}, ErrUnsupportedManifestAppendixOption
@@ -480,7 +481,7 @@ func OverwriteStoreManifest(ctx context.Context, store *NomsBlockStore, root has
 		s := tableSpec{name: h, chunkCount: c}
 		contents.specs = append(contents.specs, s)
 	}
-	contents.lock = generateLockHash(contents.root, contents.specs, contents.appendix)
+	contents.lock = generateLockHash(contents.root, contents.specs, contents.appendix, nil)
 
 	store.mm.LockForUpdate()
 	defer func() {
@@ -1307,7 +1308,7 @@ func (nbs *NomsBlockStore) updateManifest(ctx context.Context, current, last has
 	newContents := manifestContents{
 		nbfVers:  nbs.upstream.nbfVers,
 		root:     current,
-		lock:     generateLockHash(current, specs, appendixSpecs),
+		lock:     generateLockHash(current, specs, appendixSpecs, nil),
 		gcGen:    nbs.upstream.gcGen,
 		specs:    specs,
 		appendix: appendixSpecs,
@@ -1597,11 +1598,11 @@ func (nbs *NomsBlockStore) EndGC() {
 	nbs.cond.Broadcast()
 }
 
-func (nbs *NomsBlockStore) MarkAndSweepChunks(ctx context.Context, hashes <-chan []hash.Hash, dest chunks.ChunkStore) (chunks.GCFinalizer, error) {
-	return markAndSweepChunks(ctx, hashes, nbs, nbs, dest)
+func (nbs *NomsBlockStore) MarkAndSweepChunks(ctx context.Context, hashes <-chan []hash.Hash, dest chunks.ChunkStore, mode chunks.GCMode) (chunks.GCFinalizer, error) {
+	return markAndSweepChunks(ctx, hashes, nbs, nbs, dest, mode)
 }
 
-func markAndSweepChunks(ctx context.Context, hashes <-chan []hash.Hash, nbs *NomsBlockStore, src NBSCompressedChunkStore, dest chunks.ChunkStore) (chunks.GCFinalizer, error) {
+func markAndSweepChunks(ctx context.Context, hashes <-chan []hash.Hash, nbs *NomsBlockStore, src NBSCompressedChunkStore, dest chunks.ChunkStore, mode chunks.GCMode) (chunks.GCFinalizer, error) {
 	ops := nbs.SupportedOperations()
 	if !ops.CanGC || !ops.CanPrune {
 		return nil, chunks.ErrUnsupportedOperation
@@ -1611,12 +1612,20 @@ func markAndSweepChunks(ctx context.Context, hashes <-chan []hash.Hash, nbs *Nom
 		nbs.mu.RLock()
 		defer nbs.mu.RUnlock()
 
-		// check to see if the specs have changed since last gc. If they haven't bail early.
-		gcGenCheck := generateLockHash(nbs.upstream.root, nbs.upstream.specs, nbs.upstream.appendix)
+		// Check to see if the specs have changed since last gc. If they haven't bail early.
+		gcGenCheck := generateLockHash(nbs.upstream.root, nbs.upstream.specs, nbs.upstream.appendix, []byte("full"))
 		if nbs.upstream.gcGen == gcGenCheck {
+			fmt.Fprintf(color.Error, "check against full gcGen passed; nothing to collect")
 			return chunks.ErrNothingToCollect
 		}
-
+		if mode != chunks.GCMode_Full {
+			// Allow a non-full GC to match the no-op work check as well.
+			gcGenCheck := generateLockHash(nbs.upstream.root, nbs.upstream.specs, nbs.upstream.appendix, nil)
+			if nbs.upstream.gcGen == gcGenCheck {
+				fmt.Fprintf(color.Error, "check against nil gcGen passed; nothing to collect")
+				return chunks.ErrNothingToCollect
+			}
+		}
 		return nil
 	}
 	err := precheck()
@@ -1649,12 +1658,14 @@ func markAndSweepChunks(ctx context.Context, hashes <-chan []hash.Hash, nbs *Nom
 	return gcFinalizer{
 		nbs:   destNBS,
 		specs: specs,
+		mode:  mode,
 	}, nil
 }
 
 type gcFinalizer struct {
 	nbs   *NomsBlockStore
 	specs []tableSpec
+	mode  chunks.GCMode
 }
 
 func (gcf gcFinalizer) AddChunksToStore(ctx context.Context) (chunks.HasManyFunc, error) {
@@ -1670,7 +1681,7 @@ func (gcf gcFinalizer) AddChunksToStore(ctx context.Context) (chunks.HasManyFunc
 }
 
 func (gcf gcFinalizer) SwapChunksInStore(ctx context.Context) error {
-	return gcf.nbs.swapTables(ctx, gcf.specs)
+	return gcf.nbs.swapTables(ctx, gcf.specs, gcf.mode)
 }
 
 func copyMarkedChunks(ctx context.Context, keepChunks <-chan []hash.Hash, src NBSCompressedChunkStore, dest *NomsBlockStore) ([]tableSpec, error) {
@@ -1744,7 +1755,7 @@ func (nbs *NomsBlockStore) IterateAllChunks(ctx context.Context, cb func(chunk c
 	return nil
 }
 
-func (nbs *NomsBlockStore) swapTables(ctx context.Context, specs []tableSpec) (err error) {
+func (nbs *NomsBlockStore) swapTables(ctx context.Context, specs []tableSpec, mode chunks.GCMode) (err error) {
 	nbs.mu.Lock()
 	defer nbs.mu.Unlock()
 
@@ -1756,12 +1767,18 @@ func (nbs *NomsBlockStore) swapTables(ctx context.Context, specs []tableSpec) (e
 		}
 	}()
 
-	newLock := generateLockHash(nbs.upstream.root, specs, []tableSpec{})
+	newLock := generateLockHash(nbs.upstream.root, specs, []tableSpec{}, nil)
+	var extra []byte
+	if mode == chunks.GCMode_Full {
+		extra = []byte("full")
+	}
+	newGCGen := generateLockHash(nbs.upstream.root, specs, []tableSpec{}, extra)
+
 	newContents := manifestContents{
 		nbfVers: nbs.upstream.nbfVers,
 		root:    nbs.upstream.root,
 		lock:    newLock,
-		gcGen:   newLock,
+		gcGen:   newGCGen,
 		specs:   specs,
 	}
 

go/store/nbs/store_test.go

@@ -340,7 +340,7 @@ func TestNBSCopyGC(t *testing.T) {
 	go func() {
 		require.NoError(t, st.BeginGC(nil))
 		var finalizer chunks.GCFinalizer
-		finalizer, msErr = st.MarkAndSweepChunks(ctx, keepChan, nil)
+		finalizer, msErr = st.MarkAndSweepChunks(ctx, keepChan, nil, chunks.GCMode_Full)
 		if msErr == nil {
 			msErr = finalizer.SwapChunksInStore(ctx)
 		}

go/store/types/value_store.go

@@ -578,6 +578,8 @@ func (lvs *ValueStore) GC(ctx context.Context, mode GCMode, oldGenRefs, newGenRe
 	gcs, gcsOK := lvs.cs.(chunks.GenerationalCS)
 	collector, collectorOK := lvs.cs.(chunks.ChunkStoreGarbageCollector)
 
+	var chksMode chunks.GCMode
+
 	if gcsOK && collectorOK {
 		oldGen := gcs.OldGen()
 		newGen := gcs.NewGen()
@@ -586,8 +588,10 @@ func (lvs *ValueStore) GC(ctx context.Context, mode GCMode, oldGenRefs, newGenRe
 		switch mode {
 		case GCModeDefault:
 			oldGenHasMany = oldGen.HasMany
+			chksMode = chunks.GCMode_Default
 		case GCModeFull:
 			oldGenHasMany = unfilteredHashFunc
+			chksMode = chunks.GCMode_Full
 		default:
 			return fmt.Errorf("unsupported GCMode %v", mode)
 		}
@@ -617,7 +621,7 @@ func (lvs *ValueStore) GC(ctx context.Context, mode GCMode, oldGenRefs, newGenRe
 			newGenRefs.Insert(root)
 
 			var oldGenFinalizer, newGenFinalizer chunks.GCFinalizer
-			oldGenFinalizer, err = lvs.gc(ctx, oldGenRefs, oldGenHasMany, collector, oldGen, nil, func() hash.HashSet {
+			oldGenFinalizer, err = lvs.gc(ctx, oldGenRefs, oldGenHasMany, chksMode, collector, oldGen, nil, func() hash.HashSet {
 				n := lvs.transitionToNewGenGC()
 				newGenRefs.InsertAll(n)
 				return make(hash.HashSet)
@@ -638,7 +642,7 @@ func (lvs *ValueStore) GC(ctx context.Context, mode GCMode, oldGenRefs, newGenRe
 				oldGenHasMany = newFileHasMany
 			}
 
-			newGenFinalizer, err = lvs.gc(ctx, newGenRefs, oldGenHasMany, collector, newGen, safepointF, lvs.transitionToFinalizingGC)
+			newGenFinalizer, err = lvs.gc(ctx, newGenRefs, oldGenHasMany, chksMode, collector, newGen, safepointF, lvs.transitionToFinalizingGC)
 			if err != nil {
 				return err
 			}
@@ -685,7 +689,7 @@ func (lvs *ValueStore) GC(ctx context.Context, mode GCMode, oldGenRefs, newGenRe
 			newGenRefs.Insert(root)
 
 			var finalizer chunks.GCFinalizer
-			finalizer, err = lvs.gc(ctx, newGenRefs, unfilteredHashFunc, collector, collector, safepointF, lvs.transitionToFinalizingGC)
+			finalizer, err = lvs.gc(ctx, newGenRefs, unfilteredHashFunc, chunks.GCMode_Full, collector, collector, safepointF, lvs.transitionToFinalizingGC)
 			if err != nil {
 				return err
 			}
@@ -719,6 +723,7 @@ func (lvs *ValueStore) GC(ctx context.Context, mode GCMode, oldGenRefs, newGenRe
 func (lvs *ValueStore) gc(ctx context.Context,
 	toVisit hash.HashSet,
 	hashFilter chunks.HasManyFunc,
+	chksMode chunks.GCMode,
 	src, dest chunks.ChunkStoreGarbageCollector,
 	safepointF func() error,
 	finalize func() hash.HashSet) (chunks.GCFinalizer, error) {
@@ -729,7 +734,7 @@ func (lvs *ValueStore) gc(ctx context.Context,
 	eg, ctx := errgroup.WithContext(ctx)
 	eg.Go(func() error {
 		var err error
-		gcFinalizer, err = src.MarkAndSweepChunks(ctx, keepChunks, dest)
+		gcFinalizer, err = src.MarkAndSweepChunks(ctx, keepChunks, dest, chksMode)
 		return err
 	})