diff --git a/swarm/network/stream/snapshot_retrieval_test.go b/swarm/network/stream/snapshot_retrieval_test.go index 3344b80c7f..c6a187ec6a 100644 --- a/swarm/network/stream/snapshot_retrieval_test.go +++ b/swarm/network/stream/snapshot_retrieval_test.go @@ -129,6 +129,7 @@ func retrievalStreamerFunc(ctx *adapters.ServiceContext, bucket *sync.Map) (s no r := NewRegistry(addr.ID(), delivery, netStore, state.NewInmemoryStore(), &RegistryOptions{ DoSync: true, SyncUpdateDelay: 3 * time.Second, + DoRetrieve: true, }) fileStore := storage.NewFileStore(netStore, storage.NewFileStoreParams()) @@ -206,36 +207,22 @@ func runFileRetrievalTest(nodeCount int) error { for !allSuccess { for _, id := range nodeIDs { //for each expected chunk, check if it is in the local store - localChunks := conf.idToChunksMap[id] localSuccess := true - for _, ch := range localChunks { - //get the real chunk by the index in the index array - chunk := conf.hashes[ch] - log.Trace(fmt.Sprintf("node has chunk: %s:", chunk)) - //check if the expected chunk is indeed in the localstore - var err error - //check on the node's FileStore (netstore) - item, ok := sim.NodeItem(id, bucketKeyFileStore) - if !ok { - return fmt.Errorf("No registry") - } - fileStore := item.(*storage.FileStore) - //check all chunks - for i, hash := range conf.hashes { - reader, _ := fileStore.Retrieve(context.TODO(), hash) - //check that we can read the file size and that it corresponds to the generated file size - if s, err := reader.Size(ctx, nil); err != nil || s != int64(len(randomFiles[i])) { - allSuccess = false - log.Warn("Retrieve error", "err", err, "hash", hash, "nodeId", id) - } else { - log.Debug(fmt.Sprintf("File with root hash %x successfully retrieved", hash)) - } - } - if err != nil { - log.Warn(fmt.Sprintf("Chunk %s NOT found for id %s", chunk, id)) + item, ok := sim.NodeItem(id, bucketKeyFileStore) + if !ok { + return fmt.Errorf("No registry") + } + fileStore := item.(*storage.FileStore) + //check all chunks + for i, hash := range conf.hashes { + reader, _ := fileStore.Retrieve(context.TODO(), hash) + //check that we can read the file size and that it corresponds to the generated file size + if s, err := reader.Size(ctx, nil); err != nil || s != int64(len(randomFiles[i])) { + //allSuccess = false localSuccess = false + log.Warn("Retrieve error", "err", err, "hash", hash, "nodeId", id) } else { - log.Debug(fmt.Sprintf("Chunk %s IS FOUND for id %s", chunk, id)) + log.Debug(fmt.Sprintf("File with root hash %x successfully retrieved", hash)) } } allSuccess = localSuccess @@ -295,8 +282,6 @@ func runRetrievalTest(chunkCount int, nodeCount int) error { conf.addrToIDMap[string(a)] = n } - //an array for the random files - var randomFiles []string //this is the node selected for upload node := sim.RandomUpNode() item, ok := sim.NodeItem(node.ID, bucketKeyStore) @@ -318,36 +303,22 @@ func runRetrievalTest(chunkCount int, nodeCount int) error { for !allSuccess { for _, id := range nodeIDs { //for each expected chunk, check if it is in the local store - localChunks := conf.idToChunksMap[id] localSuccess := true - for _, ch := range localChunks { - //get the real chunk by the index in the index array - chunk := conf.hashes[ch] - log.Trace(fmt.Sprintf("node has chunk: %s:", chunk)) - //check if the expected chunk is indeed in the localstore - var err error - //check on the node's FileStore (netstore) - item, ok := sim.NodeItem(id, bucketKeyFileStore) - if !ok { - return fmt.Errorf("No registry") - } - fileStore := item.(*storage.FileStore) - //check all chunks - for i, hash := range conf.hashes { - reader, _ := fileStore.Retrieve(context.TODO(), hash) - //check that we can read the file size and that it corresponds to the generated file size - if s, err := reader.Size(ctx, nil); err != nil || s != int64(len(randomFiles[i])) { - allSuccess = false - log.Warn("Retrieve error", "err", err, "hash", hash, "nodeId", id) - } else { - log.Debug(fmt.Sprintf("File with root hash %x successfully retrieved", hash)) - } - } - if err != nil { - log.Warn(fmt.Sprintf("Chunk %s NOT found for id %s", chunk, id)) + //check on the node's FileStore (netstore) + item, ok := sim.NodeItem(id, bucketKeyFileStore) + if !ok { + return fmt.Errorf("No registry") + } + fileStore := item.(*storage.FileStore) + //check all chunks + for _, hash := range conf.hashes { + reader, _ := fileStore.Retrieve(context.TODO(), hash) + //check that we can read the file size and that it corresponds to the generated file size + if s, err := reader.Size(ctx, nil); err != nil || s != int64(chunkSize) { localSuccess = false + log.Warn("Retrieve error", "err", err, "hash", hash, "nodeId", id) } else { - log.Debug(fmt.Sprintf("Chunk %s IS FOUND for id %s", chunk, id)) + log.Debug(fmt.Sprintf("File with root hash %x successfully retrieved", hash)) } } allSuccess = localSuccess diff --git a/swarm/network/stream/visualized_snapshot_sync_sim_test.go b/swarm/network/stream/visualized_snapshot_sync_sim_test.go index 886afb03f8..96c0391762 100644 --- a/swarm/network/stream/visualized_snapshot_sync_sim_test.go +++ b/swarm/network/stream/visualized_snapshot_sync_sim_test.go @@ -53,7 +53,7 @@ func setupSim(serviceMap map[string]simulation.ServiceFunc) (int, int, *simulati } func watchSim(sim *simulation.Simulation) (context.Context, context.CancelFunc) { - ctx, cancelSimRun := context.WithTimeout(context.Background(), 2*time.Minute) + ctx, cancelSimRun := context.WithTimeout(context.Background(), 1*time.Minute) if _, err := sim.WaitTillHealthy(ctx, 2); err != nil { panic(err) @@ -98,10 +98,10 @@ func TestNonExistingHashesWithServer(t *testing.T) { } fileStore := item.(*storage.FileStore) fakeHash := storage.GenerateRandomChunk(1000).Address() - fmt.Println(fakeHash) + //fmt.Println(fakeHash) reader, _ := fileStore.Retrieve(context.TODO(), fakeHash) if _, err := reader.Size(ctx, nil); err != nil { - fmt.Println("expected error for non-existing chunk") + log.Debug("expected error for non-existing chunk") } time.Sleep(testDuration) @@ -109,14 +109,20 @@ func TestNonExistingHashesWithServer(t *testing.T) { return nil }) if result.Error != nil { + sendSimTerminatedEvent(sim) t.Fatal(result.Error) } + sendSimTerminatedEvent(sim) + +} + +func sendSimTerminatedEvent(sim *simulation.Simulation) { evt := &simulations.Event{ - Type: EventTypeSimTerminated, + Type: EventTypeSimTerminated, + Control: false, } sim.Net.Events().Send(evt) - } func TestSnapshotSyncWithServer(t *testing.T) { @@ -163,22 +169,25 @@ func TestSnapshotSyncWithServer(t *testing.T) { } if *e.Event.MsgCode == uint64(1) { evt := &simulations.Event{ - Type: EventTypeChunkOffered, - Node: sim.Net.GetNode(e.NodeID), + Type: EventTypeChunkOffered, + Node: sim.Net.GetNode(e.NodeID), + Control: false, //Data: fmt.Sprintf("%s", h), } sim.Net.Events().Send(evt) } else if *e.Event.MsgCode == uint64(2) { evt := &simulations.Event{ - Type: EventTypeChunkWanted, - Node: sim.Net.GetNode(e.NodeID), + Type: EventTypeChunkWanted, + Node: sim.Net.GetNode(e.NodeID), + Control: false, //Data: fmt.Sprintf("%s", h), } sim.Net.Events().Send(evt) } else if *e.Event.MsgCode == uint64(6) { evt := &simulations.Event{ - Type: EventTypeChunkDelivered, - Node: sim.Net.GetNode(e.NodeID), + Type: EventTypeChunkDelivered, + Node: sim.Net.GetNode(e.NodeID), + Control: false, //Data: fmt.Sprintf("%s", h), } sim.Net.Events().Send(evt) @@ -188,7 +197,8 @@ func TestSnapshotSyncWithServer(t *testing.T) { result := runSim(conf, ctx, sim, chunkCount) evt := &simulations.Event{ - Type: EventTypeSimTerminated, + Type: EventTypeSimTerminated, + Control: false, } sim.Net.Events().Send(evt)