check existance of reused blocks (#18)

* Add HasAttachments to RemoteProjectStore so we can query if attachment blocks actually exist
* use individual requests for compressed blob check in Jupiter
* remove weird 1000.5 to 1000.0 when converting seconds to milliseconds

1 files changed, 71 insertions, 24 deletions

diff --git a/src/zenserver/projectstore/remoteprojectstore.cpp b/src/zenserver/projectstore/remoteprojectstore.cpp
index ab207930f..d8dfb215d 100644
--- a/src/zenserver/projectstore/remoteprojectstore.cpp
+++ b/src/zenserver/projectstore/remoteprojectstore.cpp
@@ -1424,12 +1424,15 @@ UploadAttachments(WorkerThreadPool&														  WorkerPool,
 	size_t BulkAttachmentCountToUpload	= 0;
 	AttachmentsToUpload.reserve(ForceAll ? CreatedBlocks.size() + LargeAttachments.size() : Needs.size());
 
+	std::unordered_set<IoHash, IoHash::Hasher> UnknownAttachments(Needs);
+
 	for (const auto& CreatedBlock : CreatedBlocks)
 	{
 		if (ForceAll || Needs.contains(CreatedBlock.first))
 		{
 			AttachmentsToUpload.insert(CreatedBlock.first);
 			BlockAttachmentCountToUpload++;
+			UnknownAttachments.erase(CreatedBlock.first);
 		}
 	}
 	for (const IoHash& LargeAttachment : LargeAttachments)
@@ -1438,6 +1441,7 @@ UploadAttachments(WorkerThreadPool&														  WorkerPool,
 		{
 			AttachmentsToUpload.insert(LargeAttachment);
 			LargeAttachmentCountToUpload++;
+			UnknownAttachments.erase(LargeAttachment);
 		}
 	}
 	for (const std::vector<std::pair<IoHash, FetchChunkFunc>>& BlockHashes : BlockChunks)
@@ -1448,6 +1452,7 @@ UploadAttachments(WorkerThreadPool&														  WorkerPool,
 			{
 				BulkBlockAttachmentsToUpload.insert(std::make_pair(Chunk.first, Chunk.second));
 				BulkAttachmentCountToUpload++;
+				UnknownAttachments.erase(Chunk.first);
 			}
 		}
 	}
@@ -1458,6 +1463,17 @@ UploadAttachments(WorkerThreadPool&														  WorkerPool,
 		return;
 	}
 
+	if (!UnknownAttachments.empty())
+	{
+		RemoteResult.SetError(
+			gsl::narrow<int>(HttpResponseCode::NotFound),
+			fmt::format("Upload requested of {} missing attachments, the base container referenced blocks that are no longer available",
+						UnknownAttachments.size()),
+			"");
+		ReportMessage(OptionalContext, fmt::format("Aborting ({}): {}", RemoteResult.GetError(), RemoteResult.GetErrorReason()));
+		return;
+	}
+
 	if (IsCancelled(OptionalContext))
 	{
 		if (!RemoteResult.IsError())
@@ -1793,34 +1809,65 @@ SaveOplog(CidStore&				 ChunkStore,
 				ReportMessage(OptionalContext, fmt::format("Loaded oplog base container in {:.3} s", BaseContainerResult.ElapsedSeconds));
 
 				CbArrayView BlocksArray = BaseContainerResult.ContainerObject["blocks"sv].AsArrayView();
-				KnownBlocks.reserve(BlocksArray.Num());
+
+				std::vector<IoHash> BlockHashes;
+				BlockHashes.reserve(BlocksArray.Num());
 				for (CbFieldView BlockField : BlocksArray)
 				{
 					CbObjectView BlockView = BlockField.AsObjectView();
 					IoHash		 BlockHash = BlockView["rawhash"sv].AsBinaryAttachment();
+					BlockHashes.push_back(BlockHash);
+				}
 
-					std::vector<IoHash> ChunksInBlock;
-					CbArrayView			ChunksArray = BlockView["chunks"sv].AsArrayView();
-					if (BlockHash == IoHash::Zero)
+				RemoteProjectStore::HasAttachmentsResult HasResult = RemoteStore.HasAttachments(BlockHashes);
+				if (HasResult.ErrorCode == 0)
+				{
+					ReportMessage(OptionalContext,
+								  fmt::format("Checked the existance of {} block{} in remote store in {}",
+											  BlockHashes.size(),
+											  BlockHashes.size() > 1 ? "S"sv : ""sv,
+											  NiceTimeSpanMs(static_cast<uint64_t>(HasResult.ElapsedSeconds * 1000))));
+					if (HasResult.Needs.size() < BlocksArray.Num())
 					{
-						continue;
-					}
+						KnownBlocks.reserve(BlocksArray.Num() - HasResult.Needs.size());
 
-					ChunksInBlock.reserve(ChunksArray.Num());
-					for (CbFieldView ChunkField : ChunksArray)
-					{
-						ChunksInBlock.push_back(ChunkField.AsHash());
+						const std::unordered_set<IoHash, IoHash::Hasher> MissingBlocks(HasResult.Needs);
+
+						for (CbFieldView BlockField : BlocksArray)
+						{
+							CbObjectView BlockView = BlockField.AsObjectView();
+							IoHash		 BlockHash = BlockView["rawhash"sv].AsBinaryAttachment();
+							if (!MissingBlocks.contains(BlockHash))
+							{
+								std::vector<IoHash> ChunksInBlock;
+								CbArrayView			ChunksArray = BlockView["chunks"sv].AsArrayView();
+								if (BlockHash == IoHash::Zero)
+								{
+									continue;
+								}
+
+								ChunksInBlock.reserve(ChunksArray.Num());
+								for (CbFieldView ChunkField : ChunksArray)
+								{
+									ChunksInBlock.push_back(ChunkField.AsHash());
+								}
+								KnownBlocks.push_back({.BlockHash = BlockHash, .ChunksInBlock = std::move(ChunksInBlock)});
+							}
+						}
 					}
-					KnownBlocks.push_back({.BlockHash = BlockHash, .ChunksInBlock = std::move(ChunksInBlock)});
-				};
+				}
+				else
+				{
+					ReportMessage(OptionalContext,
+								  fmt::format("Unable to determine which blocks in base container exist in remote store, assuming none "
+											  "does: '{}', error code : {}",
+											  HasResult.Reason,
+											  HasResult.ErrorCode));
+				}
 			}
 		}
 	}
 
-	// TODO: We need to check if remote store actually *has* all KnownBlocks
-	// We can't reconstruct known blocks on demand as they may contain chunks that we don't have
-	// and we don't care about :(
-
 	CbObject OplogContainerObject = BuildContainer(ChunkStore,
 												   Project,
 												   Oplog,
@@ -1844,7 +1891,7 @@ SaveOplog(CidStore&				 ChunkStore,
 		if (IsCancelled(OptionalContext))
 		{
 			RemoteProjectStore::Result Result = {.ErrorCode		 = 0,
-												 .ElapsedSeconds = Timer.GetElapsedTimeMs() / 1000.500,
+												 .ElapsedSeconds = Timer.GetElapsedTimeMs() / 1000.0,
 												 .Text			 = "Operation cancelled"};
 			ReportMessage(OptionalContext, fmt::format("Aborting ({}): {}", RemoteResult.GetError(), RemoteResult.GetErrorReason()));
 			return Result;
@@ -1891,7 +1938,7 @@ SaveOplog(CidStore&				 ChunkStore,
 			if (IsCancelled(OptionalContext))
 			{
 				RemoteProjectStore::Result Result = {.ErrorCode		 = 0,
-													 .ElapsedSeconds = Timer.GetElapsedTimeMs() / 1000.500,
+													 .ElapsedSeconds = Timer.GetElapsedTimeMs() / 1000.0,
 													 .Text			 = "Operation cancelled"};
 				ReportMessage(OptionalContext, fmt::format("Aborting ({}): {}", Result.ErrorCode, Result.Text));
 				return Result;
@@ -1923,7 +1970,7 @@ SaveOplog(CidStore&				 ChunkStore,
 			if (IsCancelled(OptionalContext))
 			{
 				RemoteProjectStore::Result Result = {.ErrorCode		 = 0,
-													 .ElapsedSeconds = Timer.GetElapsedTimeMs() / 1000.500,
+													 .ElapsedSeconds = Timer.GetElapsedTimeMs() / 1000.0,
 													 .Text			 = "Operation cancelled"};
 				return Result;
 			}
@@ -1951,7 +1998,7 @@ SaveOplog(CidStore&				 ChunkStore,
 		CreatedBlocks.clear();
 	}
 	RemoteProjectStore::Result Result = RemoteResult.ConvertResult();
-	Result.ElapsedSeconds			  = Timer.GetElapsedTimeMs() / 1000.500;
+	Result.ElapsedSeconds			  = Timer.GetElapsedTimeMs() / 1000.0;
 
 	ReportMessage(OptionalContext,
 				  fmt::format("Saved oplog '{}' {} in {} ({}), Blocks: {} ({}), Attachments: {} ({})",
@@ -2077,7 +2124,7 @@ SaveOplogContainer(ProjectStore::Oplog&																 Oplog,
 		{
 			ReportMessage(OptionalContext, fmt::format("Failed to save oplog container: '{}'", "Section has unexpected data type"));
 			return RemoteProjectStore::Result{gsl::narrow<int>(HttpResponseCode::BadRequest),
-											  Timer.GetElapsedTimeMs() / 1000.500,
+											  Timer.GetElapsedTimeMs() / 1000.0,
 											  "Section has unexpected data type",
 											  "Failed to save oplog container"};
 		}
@@ -2130,7 +2177,7 @@ SaveOplogContainer(ProjectStore::Oplog&																 Oplog,
 			AppendBatch();
 		}
 	}
-	return RemoteProjectStore::Result{.ElapsedSeconds = Timer.GetElapsedTimeMs() / 1000.500};
+	return RemoteProjectStore::Result{.ElapsedSeconds = Timer.GetElapsedTimeMs() / 1000.0};
 }
 
 RemoteProjectStore::Result
@@ -2173,7 +2220,7 @@ LoadOplog(CidStore&			   ChunkStore,
 			OptionalContext,
 			fmt::format("Failed to load oplog container: '{}', error code: {}", LoadContainerResult.Reason, LoadContainerResult.ErrorCode));
 		return RemoteProjectStore::Result{.ErrorCode	  = LoadContainerResult.ErrorCode,
-										  .ElapsedSeconds = Timer.GetElapsedTimeMs() / 1000.500,
+										  .ElapsedSeconds = Timer.GetElapsedTimeMs() / 1000.0,
 										  .Reason		  = LoadContainerResult.Reason,
 										  .Text			  = LoadContainerResult.Text};
 	}
@@ -2554,7 +2601,7 @@ LoadOplog(CidStore&			   ChunkStore,
 		Result = RemoteResult.ConvertResult();
 	}
 
-	Result.ElapsedSeconds = Timer.GetElapsedTimeMs() / 1000.500;
+	Result.ElapsedSeconds = Timer.GetElapsedTimeMs() / 1000.0;
 
 	ReportMessage(OptionalContext,
 				  fmt::format("Loaded oplog '{}' {} in {} ({}), Blocks: {} ({}), Attachments: {} ({}), Stored: {} ({}), Missing: {}",