| 1 | package coreunix |
| 2 | |
| 3 | import ( |
| 4 | "bytes" |
| 5 | "context" |
| 6 | "io" |
| 7 | "math/rand" |
| 8 | "os" |
| 9 | "path/filepath" |
| 10 | "testing" |
| 11 | "time" |
| 12 | |
| 13 | "github.com/ipfs/kubo/core" |
| 14 | "github.com/ipfs/kubo/gc" |
| 15 | "github.com/ipfs/kubo/repo" |
| 16 | |
| 17 | "github.com/ipfs/boxo/blockservice" |
| 18 | blockstore "github.com/ipfs/boxo/blockstore" |
| 19 | "github.com/ipfs/boxo/files" |
| 20 | pi "github.com/ipfs/boxo/filestore/posinfo" |
| 21 | dag "github.com/ipfs/boxo/ipld/merkledag" |
| 22 | blocks "github.com/ipfs/go-block-format" |
| 23 | "github.com/ipfs/go-cid" |
| 24 | "github.com/ipfs/go-datastore" |
| 25 | syncds "github.com/ipfs/go-datastore/sync" |
| 26 | config "github.com/ipfs/kubo/config" |
| 27 | coreiface "github.com/ipfs/kubo/core/coreiface" |
| 28 | ) |
| 29 | |
| 30 | const testPeerID = "QmTFauExutTsy4XP6JbMFcw2Wa9645HJt2bTqL6qYDCKfe" |
| 31 | |
| 32 | func TestAddMultipleGCLive(t *testing.T) { |
| 33 | ctx := t.Context() |
| 34 | r := &repo.Mock{ |
| 35 | C: config.Config{ |
| 36 | Identity: config.Identity{ |
| 37 | PeerID: testPeerID, // required by offline node |
| 38 | }, |
| 39 | }, |
| 40 | D: syncds.MutexWrap(datastore.NewMapDatastore()), |
| 41 | } |
| 42 | node, err := core.NewNode(ctx, &core.BuildCfg{Repo: r}) |
| 43 | if err != nil { |
| 44 | t.Fatal(err) |
| 45 | } |
| 46 | |
| 47 | out := make(chan any, 10) |
| 48 | adder, err := NewAdder(ctx, node.Pinning, node.Blockstore, node.DAG) |
| 49 | if err != nil { |
| 50 | t.Fatal(err) |
| 51 | } |
| 52 | adder.Out = out |
| 53 | |
| 54 | // make two files with pipes so we can 'pause' the add for timing of the test |
| 55 | piper1, pipew1 := io.Pipe() |
| 56 | hangfile1 := files.NewReaderFile(piper1) |
| 57 | |
| 58 | piper2, pipew2 := io.Pipe() |
| 59 | hangfile2 := files.NewReaderFile(piper2) |
| 60 | |
| 61 | rfc := files.NewBytesFile([]byte("testfileA")) |
| 62 | |
| 63 | slf := files.NewMapDirectory(map[string]files.Node{ |
| 64 | "a": hangfile1, |
| 65 | "b": hangfile2, |
| 66 | "c": rfc, |
| 67 | }) |
| 68 | |
| 69 | go func() { |
| 70 | defer close(out) |
| 71 | _, _ = adder.AddAllAndPin(ctx, slf) |
| 72 | // Ignore errors for clarity - the real bug would be gc'ing files while adding them, not this resultant error |
| 73 | }() |
| 74 | |
| 75 | // Start writing the first file but don't close the stream |
| 76 | if _, err := pipew1.Write([]byte("some data for file a")); err != nil { |
| 77 | t.Fatal(err) |
| 78 | } |
| 79 | |
| 80 | var gc1out <-chan gc.Result |
| 81 | gc1started := make(chan struct{}) |
| 82 | go func() { |
| 83 | defer close(gc1started) |
| 84 | gc1out = gc.GC(ctx, node.Blockstore, node.Repo.Datastore(), node.Pinning, nil) |
| 85 | }() |
| 86 | |
| 87 | // Give GC goroutine time to reach GCLock (will block there waiting for adder) |
| 88 | time.Sleep(time.Millisecond * 100) |
| 89 | |
| 90 | // GC shouldn't get the lock until after the file is completely added |
| 91 | select { |
| 92 | case <-gc1started: |
| 93 | t.Fatal("gc shouldn't have started yet") |
| 94 | default: |
| 95 | } |
| 96 | |
| 97 | // finish write and unblock gc |
| 98 | pipew1.Close() |
| 99 | |
| 100 | // Wait for GC to acquire the lock |
| 101 | // The adder needs to finish processing file 'a' and call maybePauseForGC |
| 102 | // when starting file 'b' before GC can proceed |
| 103 | select { |
| 104 | case <-gc1started: |
| 105 | // GC got the lock as expected |
| 106 | case <-time.After(5 * time.Second): |
| 107 | t.Fatal("timeout waiting for GC to start - possible deadlock") |
| 108 | } |
| 109 | |
| 110 | removedHashes := make(map[string]struct{}) |
| 111 | for r := range gc1out { |
| 112 | if r.Error != nil { |
| 113 | t.Fatal(err) |
| 114 | } |
| 115 | removedHashes[r.KeyRemoved.String()] = struct{}{} |
| 116 | } |
| 117 | |
| 118 | if _, err := pipew2.Write([]byte("some data for file b")); err != nil { |
| 119 | t.Fatal(err) |
| 120 | } |
| 121 | |
| 122 | var gc2out <-chan gc.Result |
| 123 | gc2started := make(chan struct{}) |
| 124 | go func() { |
| 125 | defer close(gc2started) |
| 126 | gc2out = gc.GC(ctx, node.Blockstore, node.Repo.Datastore(), node.Pinning, nil) |
| 127 | }() |
| 128 | |
| 129 | // Give GC goroutine time to reach GCLock |
| 130 | time.Sleep(time.Millisecond * 100) |
| 131 | |
| 132 | select { |
| 133 | case <-gc2started: |
| 134 | t.Fatal("gc shouldn't have started yet") |
| 135 | default: |
| 136 | } |
| 137 | |
| 138 | pipew2.Close() |
| 139 | |
| 140 | // Wait for second GC to acquire the lock |
| 141 | // The adder needs to finish processing file 'b' and call maybePauseForGC |
| 142 | // when starting file 'c' before GC can proceed |
| 143 | select { |
| 144 | case <-gc2started: |
| 145 | // GC got the lock as expected |
| 146 | case <-time.After(5 * time.Second): |
| 147 | t.Fatal("timeout waiting for second GC to start - possible deadlock") |
| 148 | } |
| 149 | |
| 150 | for r := range gc2out { |
| 151 | if r.Error != nil { |
| 152 | t.Fatal(err) |
| 153 | } |
| 154 | removedHashes[r.KeyRemoved.String()] = struct{}{} |
| 155 | } |
| 156 | |
| 157 | for o := range out { |
| 158 | if _, ok := removedHashes[o.(*coreiface.AddEvent).Path.RootCid().String()]; ok { |
| 159 | t.Fatal("gc'ed a hash we just added") |
| 160 | } |
| 161 | } |
| 162 | } |
| 163 | |
| 164 | func TestAddGCLive(t *testing.T) { |
| 165 | ctx := t.Context() |
| 166 | r := &repo.Mock{ |
| 167 | C: config.Config{ |
| 168 | Identity: config.Identity{ |
| 169 | PeerID: testPeerID, // required by offline node |
| 170 | }, |
| 171 | }, |
| 172 | D: syncds.MutexWrap(datastore.NewMapDatastore()), |
| 173 | } |
| 174 | node, err := core.NewNode(ctx, &core.BuildCfg{Repo: r}) |
| 175 | if err != nil { |
| 176 | t.Fatal(err) |
| 177 | } |
| 178 | |
| 179 | out := make(chan any) |
| 180 | adder, err := NewAdder(ctx, node.Pinning, node.Blockstore, node.DAG) |
| 181 | if err != nil { |
| 182 | t.Fatal(err) |
| 183 | } |
| 184 | adder.Out = out |
| 185 | |
| 186 | rfa := files.NewBytesFile([]byte("testfileA")) |
| 187 | |
| 188 | // make two files with pipes so we can 'pause' the add for timing of the test |
| 189 | piper, pipew := io.Pipe() |
| 190 | hangfile := files.NewReaderFile(piper) |
| 191 | |
| 192 | rfd := files.NewBytesFile([]byte("testfileD")) |
| 193 | |
| 194 | slf := files.NewMapDirectory(map[string]files.Node{ |
| 195 | "a": rfa, |
| 196 | "b": hangfile, |
| 197 | "d": rfd, |
| 198 | }) |
| 199 | |
| 200 | addDone := make(chan struct{}) |
| 201 | go func() { |
| 202 | defer close(addDone) |
| 203 | defer close(out) |
| 204 | _, err := adder.AddAllAndPin(ctx, slf) |
| 205 | if err != nil { |
| 206 | t.Error(err) |
| 207 | } |
| 208 | }() |
| 209 | |
| 210 | addedHashes := make(map[string]struct{}) |
| 211 | select { |
| 212 | case o := <-out: |
| 213 | addedHashes[o.(*coreiface.AddEvent).Path.RootCid().String()] = struct{}{} |
| 214 | case <-addDone: |
| 215 | t.Fatal("add shouldn't complete yet") |
| 216 | } |
| 217 | |
| 218 | var gcout <-chan gc.Result |
| 219 | gcstarted := make(chan struct{}) |
| 220 | go func() { |
| 221 | defer close(gcstarted) |
| 222 | gcout = gc.GC(ctx, node.Blockstore, node.Repo.Datastore(), node.Pinning, nil) |
| 223 | }() |
| 224 | |
| 225 | // gc shouldn't start until we let the add finish its current file. |
| 226 | if _, err := pipew.Write([]byte("some data for file b")); err != nil { |
| 227 | t.Fatal(err) |
| 228 | } |
| 229 | |
| 230 | select { |
| 231 | case <-gcstarted: |
| 232 | t.Fatal("gc shouldn't have started yet") |
| 233 | default: |
| 234 | } |
| 235 | |
| 236 | time.Sleep(time.Millisecond * 100) // make sure gc gets to requesting lock |
| 237 | |
| 238 | // finish write and unblock gc |
| 239 | pipew.Close() |
| 240 | |
| 241 | // receive next object from adder |
| 242 | o := <-out |
| 243 | addedHashes[o.(*coreiface.AddEvent).Path.RootCid().String()] = struct{}{} |
| 244 | |
| 245 | <-gcstarted |
| 246 | |
| 247 | for r := range gcout { |
| 248 | if r.Error != nil { |
| 249 | t.Fatal(err) |
| 250 | } |
| 251 | if _, ok := addedHashes[r.KeyRemoved.String()]; ok { |
| 252 | t.Fatal("gc'ed a hash we just added") |
| 253 | } |
| 254 | } |
| 255 | |
| 256 | var last cid.Cid |
| 257 | for a := range out { |
| 258 | // wait for it to finish |
| 259 | c, err := cid.Decode(a.(*coreiface.AddEvent).Path.RootCid().String()) |
| 260 | if err != nil { |
| 261 | t.Fatal(err) |
| 262 | } |
| 263 | last = c |
| 264 | } |
| 265 | |
| 266 | set := cid.NewSet() |
| 267 | err = dag.Walk(ctx, dag.GetLinksWithDAG(node.DAG), last, set.Visit) |
| 268 | if err != nil { |
| 269 | t.Fatal(err) |
| 270 | } |
| 271 | } |
| 272 | |
| 273 | func testAddWPosInfo(t *testing.T, rawLeaves bool) { |
| 274 | r := &repo.Mock{ |
| 275 | C: config.Config{ |
| 276 | Identity: config.Identity{ |
| 277 | PeerID: testPeerID, // required by offline node |
| 278 | }, |
| 279 | }, |
| 280 | D: syncds.MutexWrap(datastore.NewMapDatastore()), |
| 281 | } |
| 282 | node, err := core.NewNode(context.Background(), &core.BuildCfg{Repo: r}) |
| 283 | if err != nil { |
| 284 | t.Fatal(err) |
| 285 | } |
| 286 | |
| 287 | bs := &testBlockstore{GCBlockstore: node.Blockstore, expectedPath: filepath.Join(os.TempDir(), "foo.txt"), t: t} |
| 288 | bserv := blockservice.New(bs, node.Exchange) |
| 289 | dserv := dag.NewDAGService(bserv) |
| 290 | adder, err := NewAdder(context.Background(), node.Pinning, bs, dserv) |
| 291 | if err != nil { |
| 292 | t.Fatal(err) |
| 293 | } |
| 294 | out := make(chan any) |
| 295 | adder.Out = out |
| 296 | adder.Progress = true |
| 297 | adder.RawLeaves = rawLeaves |
| 298 | adder.NoCopy = true |
| 299 | |
| 300 | data := make([]byte, 5*1024*1024) |
| 301 | rand.New(rand.NewSource(2)).Read(data) // Rand.Read never returns an error |
| 302 | fileData := io.NopCloser(bytes.NewBuffer(data)) |
| 303 | fileInfo := dummyFileInfo{"foo.txt", int64(len(data)), time.Now()} |
| 304 | file, _ := files.NewReaderPathFile(filepath.Join(os.TempDir(), "foo.txt"), fileData, &fileInfo) |
| 305 | |
| 306 | go func() { |
| 307 | defer close(adder.Out) |
| 308 | _, err = adder.AddAllAndPin(context.Background(), file) |
| 309 | if err != nil { |
| 310 | t.Error(err) |
| 311 | } |
| 312 | }() |
| 313 | for range out { |
| 314 | } |
| 315 | |
| 316 | exp := 0 |
| 317 | nonOffZero := 0 |
| 318 | if rawLeaves { |
| 319 | exp = 1 |
| 320 | nonOffZero = 19 |
| 321 | } |
| 322 | if bs.countAtOffsetZero != exp { |
| 323 | t.Fatalf("expected %d blocks with an offset at zero (one root and one leaf), got %d", exp, bs.countAtOffsetZero) |
| 324 | } |
| 325 | if bs.countAtOffsetNonZero != nonOffZero { |
| 326 | // note: the exact number will depend on the size and the sharding algo. used |
| 327 | t.Fatalf("expected %d blocks with an offset > 0, got %d", nonOffZero, bs.countAtOffsetNonZero) |
| 328 | } |
| 329 | } |
| 330 | |
| 331 | func TestAddWPosInfo(t *testing.T) { |
| 332 | testAddWPosInfo(t, false) |
| 333 | } |
| 334 | |
| 335 | func TestAddWPosInfoAndRawLeafs(t *testing.T) { |
| 336 | testAddWPosInfo(t, true) |
| 337 | } |
| 338 | |
| 339 | type testBlockstore struct { |
| 340 | blockstore.GCBlockstore |
| 341 | expectedPath string |
| 342 | t *testing.T |
| 343 | countAtOffsetZero int |
| 344 | countAtOffsetNonZero int |
| 345 | } |
| 346 | |
| 347 | func (bs *testBlockstore) Put(ctx context.Context, block blocks.Block) error { |
| 348 | bs.CheckForPosInfo(block) |
| 349 | return bs.GCBlockstore.Put(ctx, block) |
| 350 | } |
| 351 | |
| 352 | func (bs *testBlockstore) PutMany(ctx context.Context, blocks []blocks.Block) error { |
| 353 | for _, blk := range blocks { |
| 354 | bs.CheckForPosInfo(blk) |
| 355 | } |
| 356 | return bs.GCBlockstore.PutMany(ctx, blocks) |
| 357 | } |
| 358 | |
| 359 | func (bs *testBlockstore) CheckForPosInfo(block blocks.Block) { |
| 360 | fsn, ok := block.(*pi.FilestoreNode) |
| 361 | if ok { |
| 362 | posInfo := fsn.PosInfo |
| 363 | if posInfo.FullPath != bs.expectedPath { |
| 364 | bs.t.Fatal("PosInfo does not have the expected path") |
| 365 | } |
| 366 | if posInfo.Offset == 0 { |
| 367 | bs.countAtOffsetZero += 1 |
| 368 | } else { |
| 369 | bs.countAtOffsetNonZero += 1 |
| 370 | } |
| 371 | } |
| 372 | } |
| 373 | |
| 374 | type dummyFileInfo struct { |
| 375 | name string |
| 376 | size int64 |
| 377 | modTime time.Time |
| 378 | } |
| 379 | |
| 380 | func (fi *dummyFileInfo) Name() string { return fi.name } |
| 381 | func (fi *dummyFileInfo) Size() int64 { return fi.size } |
| 382 | func (fi *dummyFileInfo) Mode() os.FileMode { return 0 } |
| 383 | func (fi *dummyFileInfo) ModTime() time.Time { return fi.modTime } |
| 384 | func (fi *dummyFileInfo) IsDir() bool { return false } |
| 385 | func (fi *dummyFileInfo) Sys() any { return nil } |