Add localstack integration and DataSetDownloaderDaemon test

[kurotych]

AwsLocal is the code I use in 3 different projects to mock proto files in AWS.
So, I think it would be nice to have the core AwsLocal as workspace member in oracles repo to have capability to reuse it.

https://github.com/novalabsxyz/mobile-rewards-estimator/pull/20#issuecomment-2695429042

[Copilot]

Pull Request Overview

This PR introduces the aws_local crate for mocking S3 interactions via LocalStack, integrates it into the oracles workspace, and adds an initial dataset downloader with integration tests.

• Adds aws_local helper crate and wires it into CI
• Implements DataSetDownloaderDaemon with separate fetch_first_datasets and check_for_new_data_sets APIs
• Adds integration tests for dataset downloading in mobile_verifier

Reviewed Changes

Copilot reviewed 16 out of 16 changed files in this pull request and generated 5 comments.

Show a summary per file

File	Description
mobile_verifier/tests/integrations/boosting_oracles.rs	Adds helper functions and integration test for downloader
mobile_verifier/src/boosting_oracles/data_sets.rs	Refactors run, exposes check_for_new_data_sets, and adds fetch_first_datasets
mobile_verifier/Cargo.toml	Registers aws-local and tempfile
file_store/Cargo.toml	Switches AWS crates to workspace references
aws_local/src/lib.rs	Implements AwsLocal wrapper over LocalStack S3
aws_local/src/README.md	Adds LocalStack usage note
aws_local/Cargo.toml	Declares aws-local package
Cargo.toml	Adds aws_local workspace member and patches S3 crates
.github/workflows/{DockerCI,CI}.yml	Spins up LocalStack in CI

Comments suppressed due to low confidence (1)

mobile_verifier/tests/integrations/boosting_oracles.rs:160

• [nitpick] Function name hex_assignment_file_exist is grammatically inconsistent; consider renaming to hex_assignment_file_exists for clarity.

pub async fn hex_assignment_file_exist(pool: &PgPool, filename: &str) -> bool {

[Copilot]

The run method no longer calls check_for_new_data_sets, so it only fetches initial datasets but never polls for new updates. Consider invoking self.check_for_new_data_sets().await? (or a looping mechanism) within run to restore the intended behavior.

Suggested change

	self.fetch_first_datasets().await?;
	self.fetch_first_datasets().await?;
	self.check_for_new_data_sets().await?;

[Copilot]

Calling endpoint.unwrap() will panic if settings.endpoint is None. Consider handling the Option safely (e.g., using if let Some(ep) = endpoint or returning an error with a clear message).

Suggested change

	.unwrap(),
	_ => None,
	};
	let region = Region::new(settings.region.clone());
	let region_provider = RegionProviderChain::first_try(region).or_default_provider();
	let mut config = aws_config::from_env().region(region_provider);
	config = config.endpoint_resolver(endpoint.unwrap());
	.map_err(|e| anyhow!("Invalid endpoint URI: {}", e))?,
	None => return Err(anyhow!("AWS endpoint is not set in the settings")),
	};
	let region = Region::new(settings.region.clone());
	let region_provider = RegionProviderChain::first_try(region).or_default_provider();
	let mut config = aws_config::from_env().region(region_provider);
	if let Some(endpoint) = endpoint {
	config = config.endpoint_resolver(endpoint);
	}

[Copilot]

The loop waits for directory emptiness but never breaks on timeout expiry, risking an infinite loop. Consider checking timeout and breaking or returning an error when elapsed.

Suggested change

	if is_dir_has_files(&dir_path_clone) {
	if is_dir_has_files(&dir_path_clone) {
	if timeout <= std::time::Duration::ZERO {
	eprintln!("Timeout expired while waiting for directory to become empty.");
	break;
	}

[kurotych]

timeout -= dur;
Causes panic and breaks loop if timeout happens

[Copilot]

Using assert! inside a spawned task can cause panics that aren’t observed by the main test harness. Consider returning errors through a channel or awaiting the task’s result to propagate failures.

Suggested change

	tokio::spawn(async move {
	let uploaded_files = item_recv.await.unwrap().unwrap();
	assert!(uploaded_files.len() == 1);
	let (tx, rx) = tokio::sync::oneshot::channel();
	tokio::spawn(async move {
	let uploaded_files = item_recv.await.unwrap().unwrap();
	if uploaded_files.len() != 1 {
	let _ = tx.send(Err(anyhow!("Expected exactly one uploaded file, but found {}", uploaded_files.len())));
	return;
	}

[Copilot]

[nitpick] Returning a tuple (DataSetDownloaderDaemon, PathBuf, String) can be unclear to callers. Consider defining a small struct to name each field for better readability and maintainability.

Suggested change

	pub async fn create_data_set_downloader(
	pool: PgPool,
	file_paths: Vec<PathBuf>,
	file_upload: FileUpload,
	new_coverage_object_notification: NewCoverageObjectNotification,
	tmp_dir: &TempDir,
	) -> (DataSetDownloaderDaemon, PathBuf, String) {
	pub struct DataSetDownloaderResult {
	pub downloader: DataSetDownloaderDaemon,
	pub data_set_directory: PathBuf,
	pub bucket_name: String,
	}
	pub async fn create_data_set_downloader(
	pool: PgPool,
	file_paths: Vec<PathBuf>,
	file_upload: FileUpload,
	new_coverage_object_notification: NewCoverageObjectNotification,
	tmp_dir: &TempDir,
	) -> DataSetDownloaderResult {

[michaeldjeffrey]

Can we use tempdir here as well?
We don't need to do any cleanup ourselves. All files are removed when the TempDir is dropped.

In the PR we first encountered AwsLocal (https://github.com/novalabsxyz/mobile-rewards-estimator/pull/20), this function was never used. Is it in fact not used in any of your projects and we can drop it?

[kurotych]

@michaeldjeffrey
This function is used at least in the CAS. It serves as the basis for functions that want to put a particular type of proto file.

https://github.com/novalabsxyz/mobile-coverage/blob/master/cas/tests/common/aws_mock.rs#L141

Can we use tempdir here as well?

Yes, changed