A fork of attic a self-hostable Nix Binary Cache server
0
fork

Configure Feed

Select the types of activity you want to include in your feed.

server/upload_path: Create chunkrefs during the upload

+69 -38
+69 -38
server/src/api/v1/upload_path.rs
··· 296 296 297 297 let nar_size_db = i64::try_from(upload_info.nar_size).map_err(ServerError::request_error)?; 298 298 299 + // Create a pending NAR entry 300 + let nar_id = { 301 + let model = nar::ActiveModel { 302 + state: Set(NarState::PendingUpload), 303 + compression: Set(compression.to_string()), 304 + 305 + nar_hash: Set(upload_info.nar_hash.to_typed_base16()), 306 + nar_size: Set(nar_size_db), 307 + 308 + num_chunks: Set(0), 309 + 310 + created_at: Set(Utc::now()), 311 + ..Default::default() 312 + }; 313 + 314 + let insertion = Nar::insert(model) 315 + .exec(database) 316 + .await 317 + .map_err(ServerError::database_error)?; 318 + 319 + insertion.last_insert_id 320 + }; 321 + 322 + let cleanup = Finally::new({ 323 + let database = database.clone(); 324 + let nar_model = nar::ActiveModel { 325 + id: Set(nar_id), 326 + ..Default::default() 327 + }; 328 + 329 + async move { 330 + tracing::warn!("Error occurred - Cleaning up NAR entry"); 331 + 332 + if let Err(e) = Nar::delete(nar_model).exec(&database).await { 333 + tracing::warn!("Failed to unregister failed NAR: {}", e); 334 + } 335 + } 336 + }); 337 + 299 338 // FIXME: Maybe the client will send much more data than claimed 300 339 let (stream, nar_compute) = StreamHasher::new(stream, Sha256::new()); 301 340 let mut chunks = chunk_stream( ··· 308 347 let upload_chunk_limit = Arc::new(Semaphore::new(CONCURRENT_CHUNK_UPLOADS)); 309 348 let mut futures = Vec::new(); 310 349 350 + let mut chunk_idx = 0; 311 351 while let Some(bytes) = chunks.next().await { 312 352 let bytes = bytes.map_err(ServerError::request_error)?; 313 353 let data = ChunkData::Bytes(bytes); ··· 327 367 data, 328 368 compression_type, 329 369 compression_level, 330 - database, 370 + database.clone(), 331 371 state, 332 372 require_proof_of_possession, 333 373 ) 334 374 .await?; 375 + 376 + // Create mapping from the NAR to the chunk 377 + ChunkRef::insert(chunkref::ActiveModel { 378 + nar_id: Set(nar_id), 379 + seq: Set(chunk_idx as i32), 380 + chunk_id: Set(Some(chunk.guard.id)), 381 + chunk_hash: Set(chunk.guard.chunk_hash.clone()), 382 + compression: Set(chunk.guard.compression.clone()), 383 + ..Default::default() 384 + }) 385 + .exec(&database) 386 + .await 387 + .map_err(ServerError::database_error)?; 388 + 335 389 drop(permit); 336 390 Ok(chunk) 337 391 }) 338 392 }); 393 + 394 + chunk_idx += 1; 339 395 } 340 396 341 397 // Confirm that the NAR Hash and Size are correct ··· 374 430 .await 375 431 .map_err(ServerError::database_error)?; 376 432 377 - // Create a NAR entry 378 - let nar_id = { 379 - let model = nar::ActiveModel { 380 - state: Set(NarState::Valid), 381 - compression: Set(compression.to_string()), 382 - 383 - nar_hash: Set(upload_info.nar_hash.to_typed_base16()), 384 - nar_size: Set(nar_size_db), 385 - 386 - num_chunks: Set(chunks.len() as i32), 387 - 388 - created_at: Set(Utc::now()), 389 - ..Default::default() 390 - }; 391 - 392 - let insertion = Nar::insert(model) 393 - .exec(&txn) 394 - .await 395 - .map_err(ServerError::database_error)?; 396 - 397 - insertion.last_insert_id 398 - }; 399 - 400 - // Create mappings from the NAR to the chunks 401 - for (i, chunk) in chunks.iter().enumerate() { 402 - ChunkRef::insert(chunkref::ActiveModel { 403 - nar_id: Set(nar_id), 404 - seq: Set(i as i32), 405 - chunk_id: Set(Some(chunk.guard.id)), 406 - chunk_hash: Set(chunk.guard.chunk_hash.clone()), 407 - compression: Set(chunk.guard.compression.clone()), 408 - ..Default::default() 409 - }) 410 - .exec(&txn) 411 - .await 412 - .map_err(ServerError::database_error)?; 413 - } 433 + // Set num_chunks and mark the NAR as Valid 434 + Nar::update(nar::ActiveModel { 435 + id: Set(nar_id), 436 + state: Set(NarState::Valid), 437 + num_chunks: Set(chunks.len() as i32), 438 + ..Default::default() 439 + }) 440 + .exec(&txn) 441 + .await 442 + .map_err(ServerError::database_error)?; 414 443 415 444 // Create a mapping granting the local cache access to the NAR 416 445 Object::delete_many() ··· 432 461 .map_err(ServerError::database_error)?; 433 462 434 463 txn.commit().await.map_err(ServerError::database_error)?; 464 + 465 + cleanup.cancel(); 435 466 436 467 Ok(Json(UploadPathResult { 437 468 kind: UploadPathResultKind::Uploaded,