Skip to content

go/store/nbs: archive_chunk_source.go: Make sure we skip getRecords marked found when fetching chunks. #9317

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
reltuk merged 1 commit into from
Jun 6, 2025
Merged

go/store/nbs: archive_chunk_source.go: Make sure we skip getRecords marked found when fetching chunks. #9317

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions go/store/nbs/archive_chunk_source.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -212,6 +212,9 @@ func (acs archiveChunkSource) getRecordRanges(_ context.Context, requests []getR
func (acs archiveChunkSource) getManyCompressed(ctx context.Context, eg *errgroup.Group, reqs []getRecord, found func(context.Context, ToChunker), keeper keeperF, stats *Stats) (bool, gcBehavior, error) {
foundAll := true
for i, req := range reqs {
if req.found {
continue
}
h := *req.a
toChk, err := acs.aRdr.getAsToChunker(ctx, h, stats)
if err != nil {
Expand Down
85 changes: 57 additions & 28 deletions integration-tests/go-sql-server-driver/auto_gc_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,36 +34,52 @@ import (
func TestAutoGC(t *testing.T) {
t.Parallel()
var enabled_16, final_16, disabled, final_disabled RepoSize
numStatements, numCommits := 512, 16
if testing.Short() || os.Getenv("CI") != "" {
numStatements = 64
}
t.Run("Enable", func(t *testing.T) {
t.Parallel()
t.Run("CommitEvery16", func(t *testing.T) {
t.Parallel()
var s AutoGCTest
s.Enable = true
enabled_16, final_16 = runAutoGCTest(t, &s, 64, 16)
assert.Contains(t, string(s.PrimaryServer.Output.Bytes()), "Successfully completed auto GC")
t.Logf("repo size before final gc: %v", enabled_16)
t.Logf("repo size after final gc: %v", final_16)
})
t.Run("ClusterReplication", func(t *testing.T) {
t.Parallel()
var s AutoGCTest
s.Enable = true
s.Replicate = true
enabled_16, final_16 = runAutoGCTest(t, &s, 64, 16)
assert.Contains(t, string(s.PrimaryServer.Output.Bytes()), "Successfully completed auto GC")
assert.Contains(t, string(s.StandbyServer.Output.Bytes()), "Successfully completed auto GC")
t.Logf("repo size before final gc: %v", enabled_16)
t.Logf("repo size after final gc: %v", final_16)
rs, err := GetRepoSize(s.StandbyDir)
require.NoError(t, err)
t.Logf("standby size: %v", rs)
})
for _, sa := range []struct {
archive bool
name string
}{{true, "Archive"}, {false, "NoArchive"}} {
t.Run(sa.name, func(t *testing.T) {
t.Run("CommitEvery16", func(t *testing.T) {
t.Parallel()
var s AutoGCTest
s.Enable = true
s.Archive = sa.archive
enabled_16, final_16 = runAutoGCTest(t, &s, numStatements, numCommits)
assert.Contains(t, string(s.PrimaryServer.Output.Bytes()), "Successfully completed auto GC")
assert.NotContains(t, string(s.PrimaryServer.Output.Bytes()), "dangling references requested during GC")
t.Logf("repo size before final gc: %v", enabled_16)
t.Logf("repo size after final gc: %v", final_16)
})
t.Run("ClusterReplication", func(t *testing.T) {
t.Parallel()
var s AutoGCTest
s.Enable = true
s.Replicate = true
s.Archive = sa.archive
enabled_16, final_16 = runAutoGCTest(t, &s, numStatements, numCommits)
assert.Contains(t, string(s.PrimaryServer.Output.Bytes()), "Successfully completed auto GC")
assert.Contains(t, string(s.StandbyServer.Output.Bytes()), "Successfully completed auto GC")
assert.NotContains(t, string(s.PrimaryServer.Output.Bytes()), "dangling references requested during GC")
assert.NotContains(t, string(s.StandbyServer.Output.Bytes()), "dangling references requested during GC")
t.Logf("repo size before final gc: %v", enabled_16)
t.Logf("repo size after final gc: %v", final_16)
rs, err := GetRepoSize(s.StandbyDir)
require.NoError(t, err)
t.Logf("standby size: %v", rs)
})
})
}
})
t.Run("Disabled", func(t *testing.T) {
t.Parallel()
var s AutoGCTest
disabled, final_disabled = runAutoGCTest(t, &s, 64, 128)
disabled, final_disabled = runAutoGCTest(t, &s, numStatements, 128)
assert.NotContains(t, string(s.PrimaryServer.Output.Bytes()), "Successfully completed auto GC")
t.Logf("repo size before final gc: %v", disabled)
t.Logf("repo size after final gc: %v", final_disabled)
Expand All @@ -77,6 +93,7 @@ func TestAutoGC(t *testing.T) {

type AutoGCTest struct {
Enable bool
Archive bool
PrimaryDir string
PrimaryServer *driver.SqlServer
PrimaryDB *sql.DB
Expand Down Expand Up @@ -121,13 +138,19 @@ func (s *AutoGCTest) CreatePrimaryServer(ctx context.Context, t *testing.T, u dr
repo, err := rs.MakeRepo("auto_gc_test")
require.NoError(t, err)

archiveFragment := ``
if s.Archive {
archiveFragment = `
archive_level: 1`
}

behaviorFragment := fmt.Sprintf(`
behavior:
auto_gc_behavior:
enable: %v
enable: %v%v
listener:
port: {{get_port "primary_server_port"}}
`, s.Enable)
`, s.Enable, archiveFragment)

var clusterFragment string
if s.Replicate {
Expand Down Expand Up @@ -174,14 +197,20 @@ func (s *AutoGCTest) CreateStandbyServer(ctx context.Context, t *testing.T, u dr
repo, err := rs.MakeRepo("auto_gc_test")
require.NoError(t, err)

archiveFragment := ``
if s.Archive {
archiveFragment = `
archive_level: 1`
}

behaviorFragment := fmt.Sprintf(`
listener:
host: 0.0.0.0
port: {{get_port "standby_server_port"}}
behavior:
auto_gc_behavior:
enable: %v
`, s.Enable)
enable: %v%v
`, s.Enable, archiveFragment)

var clusterFragment string
if s.Replicate {
Expand Down
Loading