| 1 | // Requires Go 1.25+ for testing/synctest. |
| 2 | //go:build go1.25 |
| 3 | |
| 4 | package commands |
| 5 | |
| 6 | // This file contains unit tests for the --heal-timeout flag functionality |
| 7 | // using testing/synctest to avoid waiting for real timeouts. |
| 8 | // |
| 9 | // End-to-end tests for the full 'ipfs repo verify' command (including --drop |
| 10 | // and --heal flags) are located in test/cli/repo_verify_test.go. |
| 11 | |
| 12 | import ( |
| 13 | "bytes" |
| 14 | "context" |
| 15 | "errors" |
| 16 | "io" |
| 17 | "sync" |
| 18 | "testing" |
| 19 | "testing/synctest" |
| 20 | "time" |
| 21 | |
| 22 | blocks "github.com/ipfs/go-block-format" |
| 23 | "github.com/ipfs/go-cid" |
| 24 | ipld "github.com/ipfs/go-ipld-format" |
| 25 | coreiface "github.com/ipfs/kubo/core/coreiface" |
| 26 | "github.com/ipfs/kubo/core/coreiface/options" |
| 27 | "github.com/stretchr/testify/assert" |
| 28 | "github.com/stretchr/testify/require" |
| 29 | |
| 30 | "github.com/ipfs/boxo/path" |
| 31 | ) |
| 32 | |
| 33 | func TestVerifyWorkerHealTimeout(t *testing.T) { |
| 34 | t.Run("heal succeeds before timeout", func(t *testing.T) { |
| 35 | synctest.Test(t, func(t *testing.T) { |
| 36 | const healTimeout = 5 * time.Second |
| 37 | testCID := cid.MustParse("bafybeigdyrzt5sfp7udm7hu76uh7y26nf3efuylqabf3oclgtqy55fbzdi") |
| 38 | |
| 39 | // Setup channels |
| 40 | keys := make(chan cid.Cid, 1) |
| 41 | keys <- testCID |
| 42 | close(keys) |
| 43 | results := make(chan *verifyResult, 1) |
| 44 | |
| 45 | // Mock blockstore that returns error (simulating corruption) |
| 46 | mockBS := &mockBlockstore{ |
| 47 | getError: errors.New("corrupt block"), |
| 48 | } |
| 49 | |
| 50 | // Mock API where Block().Get() completes before timeout |
| 51 | mockAPI := &mockCoreAPI{ |
| 52 | blockAPI: &mockBlockAPI{ |
| 53 | getDelay: 2 * time.Second, // Less than healTimeout |
| 54 | data: []byte("healed data"), |
| 55 | }, |
| 56 | } |
| 57 | |
| 58 | var wg sync.WaitGroup |
| 59 | wg.Add(1) |
| 60 | |
| 61 | // Run worker |
| 62 | go verifyWorkerRun(t.Context(), &wg, keys, results, mockBS, mockAPI, true, true, healTimeout) |
| 63 | |
| 64 | // Advance time past the mock delay but before timeout |
| 65 | time.Sleep(3 * time.Second) |
| 66 | synctest.Wait() |
| 67 | |
| 68 | wg.Wait() |
| 69 | close(results) |
| 70 | |
| 71 | // Verify heal succeeded |
| 72 | result := <-results |
| 73 | require.NotNil(t, result) |
| 74 | assert.Equal(t, verifyStateCorruptHealed, result.state) |
| 75 | assert.Contains(t, result.msg, "healed") |
| 76 | }) |
| 77 | }) |
| 78 | |
| 79 | t.Run("heal fails due to timeout", func(t *testing.T) { |
| 80 | synctest.Test(t, func(t *testing.T) { |
| 81 | const healTimeout = 2 * time.Second |
| 82 | testCID := cid.MustParse("bafybeigdyrzt5sfp7udm7hu76uh7y26nf3efuylqabf3oclgtqy55fbzdi") |
| 83 | |
| 84 | // Setup channels |
| 85 | keys := make(chan cid.Cid, 1) |
| 86 | keys <- testCID |
| 87 | close(keys) |
| 88 | results := make(chan *verifyResult, 1) |
| 89 | |
| 90 | // Mock blockstore that returns error (simulating corruption) |
| 91 | mockBS := &mockBlockstore{ |
| 92 | getError: errors.New("corrupt block"), |
| 93 | } |
| 94 | |
| 95 | // Mock API where Block().Get() takes longer than healTimeout |
| 96 | mockAPI := &mockCoreAPI{ |
| 97 | blockAPI: &mockBlockAPI{ |
| 98 | getDelay: 5 * time.Second, // More than healTimeout |
| 99 | data: []byte("healed data"), |
| 100 | }, |
| 101 | } |
| 102 | |
| 103 | var wg sync.WaitGroup |
| 104 | wg.Add(1) |
| 105 | |
| 106 | // Run worker |
| 107 | go verifyWorkerRun(t.Context(), &wg, keys, results, mockBS, mockAPI, true, true, healTimeout) |
| 108 | |
| 109 | // Advance time past timeout |
| 110 | time.Sleep(3 * time.Second) |
| 111 | synctest.Wait() |
| 112 | |
| 113 | wg.Wait() |
| 114 | close(results) |
| 115 | |
| 116 | // Verify heal failed due to timeout |
| 117 | result := <-results |
| 118 | require.NotNil(t, result) |
| 119 | assert.Equal(t, verifyStateCorruptHealFailed, result.state) |
| 120 | assert.Contains(t, result.msg, "failed to heal") |
| 121 | assert.Contains(t, result.msg, "context deadline exceeded") |
| 122 | }) |
| 123 | }) |
| 124 | |
| 125 | t.Run("heal with zero timeout still attempts heal", func(t *testing.T) { |
| 126 | synctest.Test(t, func(t *testing.T) { |
| 127 | const healTimeout = 0 // Zero timeout means no timeout |
| 128 | testCID := cid.MustParse("bafybeigdyrzt5sfp7udm7hu76uh7y26nf3efuylqabf3oclgtqy55fbzdi") |
| 129 | |
| 130 | // Setup channels |
| 131 | keys := make(chan cid.Cid, 1) |
| 132 | keys <- testCID |
| 133 | close(keys) |
| 134 | results := make(chan *verifyResult, 1) |
| 135 | |
| 136 | // Mock blockstore that returns error (simulating corruption) |
| 137 | mockBS := &mockBlockstore{ |
| 138 | getError: errors.New("corrupt block"), |
| 139 | } |
| 140 | |
| 141 | // Mock API that succeeds quickly |
| 142 | mockAPI := &mockCoreAPI{ |
| 143 | blockAPI: &mockBlockAPI{ |
| 144 | getDelay: 100 * time.Millisecond, |
| 145 | data: []byte("healed data"), |
| 146 | }, |
| 147 | } |
| 148 | |
| 149 | var wg sync.WaitGroup |
| 150 | wg.Add(1) |
| 151 | |
| 152 | // Run worker |
| 153 | go verifyWorkerRun(t.Context(), &wg, keys, results, mockBS, mockAPI, true, true, healTimeout) |
| 154 | |
| 155 | // Advance time to let heal complete |
| 156 | time.Sleep(200 * time.Millisecond) |
| 157 | synctest.Wait() |
| 158 | |
| 159 | wg.Wait() |
| 160 | close(results) |
| 161 | |
| 162 | // Verify heal succeeded even with zero timeout |
| 163 | result := <-results |
| 164 | require.NotNil(t, result) |
| 165 | assert.Equal(t, verifyStateCorruptHealed, result.state) |
| 166 | assert.Contains(t, result.msg, "healed") |
| 167 | }) |
| 168 | }) |
| 169 | |
| 170 | t.Run("multiple blocks with different timeout outcomes", func(t *testing.T) { |
| 171 | synctest.Test(t, func(t *testing.T) { |
| 172 | const healTimeout = 3 * time.Second |
| 173 | testCID1 := cid.MustParse("bafybeigdyrzt5sfp7udm7hu76uh7y26nf3efuylqabf3oclgtqy55fbzdi") |
| 174 | testCID2 := cid.MustParse("bafybeihvvulpp4evxj7x7armbqcyg6uezzuig6jp3lktpbovlqfkjtgyby") |
| 175 | |
| 176 | // Setup channels |
| 177 | keys := make(chan cid.Cid, 2) |
| 178 | keys <- testCID1 |
| 179 | keys <- testCID2 |
| 180 | close(keys) |
| 181 | results := make(chan *verifyResult, 2) |
| 182 | |
| 183 | // Mock blockstore that always returns error (all blocks corrupt) |
| 184 | mockBS := &mockBlockstore{ |
| 185 | getError: errors.New("corrupt block"), |
| 186 | } |
| 187 | |
| 188 | // Create two mock block APIs with different delays |
| 189 | // We'll need to alternate which one gets used |
| 190 | // For simplicity, use one that succeeds fast |
| 191 | mockAPI := &mockCoreAPI{ |
| 192 | blockAPI: &mockBlockAPI{ |
| 193 | getDelay: 1 * time.Second, // Less than healTimeout - will succeed |
| 194 | data: []byte("healed data"), |
| 195 | }, |
| 196 | } |
| 197 | |
| 198 | var wg sync.WaitGroup |
| 199 | wg.Add(2) // Two workers |
| 200 | |
| 201 | // Run two workers |
| 202 | go verifyWorkerRun(t.Context(), &wg, keys, results, mockBS, mockAPI, true, true, healTimeout) |
| 203 | go verifyWorkerRun(t.Context(), &wg, keys, results, mockBS, mockAPI, true, true, healTimeout) |
| 204 | |
| 205 | // Advance time to let both complete |
| 206 | time.Sleep(2 * time.Second) |
| 207 | synctest.Wait() |
| 208 | |
| 209 | wg.Wait() |
| 210 | close(results) |
| 211 | |
| 212 | // Collect results |
| 213 | var healedCount int |
| 214 | for result := range results { |
| 215 | if result.state == verifyStateCorruptHealed { |
| 216 | healedCount++ |
| 217 | } |
| 218 | } |
| 219 | |
| 220 | // Both should heal successfully (both under timeout) |
| 221 | assert.Equal(t, 2, healedCount) |
| 222 | }) |
| 223 | }) |
| 224 | |
| 225 | t.Run("valid block is not healed", func(t *testing.T) { |
| 226 | synctest.Test(t, func(t *testing.T) { |
| 227 | const healTimeout = 5 * time.Second |
| 228 | testCID := cid.MustParse("bafybeigdyrzt5sfp7udm7hu76uh7y26nf3efuylqabf3oclgtqy55fbzdi") |
| 229 | |
| 230 | // Setup channels |
| 231 | keys := make(chan cid.Cid, 1) |
| 232 | keys <- testCID |
| 233 | close(keys) |
| 234 | results := make(chan *verifyResult, 1) |
| 235 | |
| 236 | // Mock blockstore that returns valid block (no error) |
| 237 | mockBS := &mockBlockstore{ |
| 238 | block: blocks.NewBlock([]byte("valid data")), |
| 239 | } |
| 240 | |
| 241 | // Mock API (won't be called since block is valid) |
| 242 | mockAPI := &mockCoreAPI{ |
| 243 | blockAPI: &mockBlockAPI{}, |
| 244 | } |
| 245 | |
| 246 | var wg sync.WaitGroup |
| 247 | wg.Add(1) |
| 248 | |
| 249 | // Run worker with heal enabled |
| 250 | go verifyWorkerRun(t.Context(), &wg, keys, results, mockBS, mockAPI, false, true, healTimeout) |
| 251 | |
| 252 | synctest.Wait() |
| 253 | |
| 254 | wg.Wait() |
| 255 | close(results) |
| 256 | |
| 257 | // Verify block is marked valid, not healed |
| 258 | result := <-results |
| 259 | require.NotNil(t, result) |
| 260 | assert.Equal(t, verifyStateValid, result.state) |
| 261 | assert.Empty(t, result.msg) |
| 262 | }) |
| 263 | }) |
| 264 | } |
| 265 | |
| 266 | // mockBlockstore implements a minimal blockstore for testing |
| 267 | type mockBlockstore struct { |
| 268 | getError error |
| 269 | block blocks.Block |
| 270 | } |
| 271 | |
| 272 | func (m *mockBlockstore) Get(ctx context.Context, c cid.Cid) (blocks.Block, error) { |
| 273 | if m.getError != nil { |
| 274 | return nil, m.getError |
| 275 | } |
| 276 | return m.block, nil |
| 277 | } |
| 278 | |
| 279 | func (m *mockBlockstore) DeleteBlock(ctx context.Context, c cid.Cid) error { |
| 280 | return nil |
| 281 | } |
| 282 | |
| 283 | func (m *mockBlockstore) Has(ctx context.Context, c cid.Cid) (bool, error) { |
| 284 | return m.block != nil, nil |
| 285 | } |
| 286 | |
| 287 | func (m *mockBlockstore) GetSize(ctx context.Context, c cid.Cid) (int, error) { |
| 288 | if m.block != nil { |
| 289 | return len(m.block.RawData()), nil |
| 290 | } |
| 291 | return 0, errors.New("block not found") |
| 292 | } |
| 293 | |
| 294 | func (m *mockBlockstore) Put(ctx context.Context, b blocks.Block) error { |
| 295 | return nil |
| 296 | } |
| 297 | |
| 298 | func (m *mockBlockstore) PutMany(ctx context.Context, bs []blocks.Block) error { |
| 299 | return nil |
| 300 | } |
| 301 | |
| 302 | func (m *mockBlockstore) AllKeysChan(ctx context.Context) (<-chan cid.Cid, error) { |
| 303 | return nil, errors.New("not implemented") |
| 304 | } |
| 305 | |
| 306 | func (m *mockBlockstore) HashOnRead(enabled bool) { |
| 307 | } |
| 308 | |
| 309 | // mockBlockAPI implements BlockAPI for testing |
| 310 | type mockBlockAPI struct { |
| 311 | getDelay time.Duration |
| 312 | getError error |
| 313 | data []byte |
| 314 | } |
| 315 | |
| 316 | func (m *mockBlockAPI) Get(ctx context.Context, p path.Path) (io.Reader, error) { |
| 317 | if m.getDelay > 0 { |
| 318 | select { |
| 319 | case <-time.After(m.getDelay): |
| 320 | // Delay completed |
| 321 | case <-ctx.Done(): |
| 322 | return nil, ctx.Err() |
| 323 | } |
| 324 | } |
| 325 | if m.getError != nil { |
| 326 | return nil, m.getError |
| 327 | } |
| 328 | return bytes.NewReader(m.data), nil |
| 329 | } |
| 330 | |
| 331 | func (m *mockBlockAPI) Put(ctx context.Context, r io.Reader, opts ...options.BlockPutOption) (coreiface.BlockStat, error) { |
| 332 | return nil, errors.New("not implemented") |
| 333 | } |
| 334 | |
| 335 | func (m *mockBlockAPI) Rm(ctx context.Context, p path.Path, opts ...options.BlockRmOption) error { |
| 336 | return errors.New("not implemented") |
| 337 | } |
| 338 | |
| 339 | func (m *mockBlockAPI) Stat(ctx context.Context, p path.Path) (coreiface.BlockStat, error) { |
| 340 | return nil, errors.New("not implemented") |
| 341 | } |
| 342 | |
| 343 | // mockCoreAPI implements minimal CoreAPI for testing |
| 344 | type mockCoreAPI struct { |
| 345 | blockAPI *mockBlockAPI |
| 346 | } |
| 347 | |
| 348 | func (m *mockCoreAPI) Block() coreiface.BlockAPI { |
| 349 | return m.blockAPI |
| 350 | } |
| 351 | |
| 352 | func (m *mockCoreAPI) Unixfs() coreiface.UnixfsAPI { return nil } |
| 353 | func (m *mockCoreAPI) Dag() coreiface.APIDagService { return nil } |
| 354 | func (m *mockCoreAPI) Name() coreiface.NameAPI { return nil } |
| 355 | func (m *mockCoreAPI) Key() coreiface.KeyAPI { return nil } |
| 356 | func (m *mockCoreAPI) Pin() coreiface.PinAPI { return nil } |
| 357 | func (m *mockCoreAPI) Object() coreiface.ObjectAPI { return nil } |
| 358 | func (m *mockCoreAPI) Swarm() coreiface.SwarmAPI { return nil } |
| 359 | func (m *mockCoreAPI) PubSub() coreiface.PubSubAPI { return nil } |
| 360 | func (m *mockCoreAPI) Routing() coreiface.RoutingAPI { return nil } |
| 361 | |
| 362 | func (m *mockCoreAPI) ResolvePath(ctx context.Context, p path.Path) (path.ImmutablePath, []string, error) { |
| 363 | return path.ImmutablePath{}, nil, errors.New("not implemented") |
| 364 | } |
| 365 | |
| 366 | func (m *mockCoreAPI) ResolveNode(ctx context.Context, p path.Path) (ipld.Node, error) { |
| 367 | return nil, errors.New("not implemented") |
| 368 | } |
| 369 | |
| 370 | func (m *mockCoreAPI) WithOptions(...options.ApiOption) (coreiface.CoreAPI, error) { |
| 371 | return nil, errors.New("not implemented") |
| 372 | } |