import datetime
import os
import json
import shlex
import subprocess
from typing import List, Tuple, Dict, Any, Optional
import time
from celery.utils.log import get_task_logger
from sqlalchemy import and_, or_
from sqlalchemy.orm import Session
from ccat_ops_db import models
from .database import DatabaseConnection
from .setup_celery_app import app, make_celery_task
from .utils import (
create_local_folder,
create_remote_folder,
make_bbcp_command,
parse_bbcp_output,
safe_join,
SSH_OPTS,
)
from .config.config import ccat_data_transfer_settings
from .operational_config import OperationalConfig
from .decorators import track_metrics
from .metrics import HousekeepingMetrics
from .logging_utils import BBCPLogHandler
from .exceptions import (
BBCPError,
NetworkError,
SegmentationFaultError,
DestinationFileExistsError,
ArchiveCorruptionError,
)
from .boundary import ensure_readable, diagnose_source_missing
from .logging_utils import get_structured_logger
from .bbcp_settings import BBCPSettings
from .notification_service import NotificationClient
from .buffer_manager import buffer_manager
from .utils import get_redis_connection
from .queue_discovery import route_task_by_location
from .operation_types import OperationType
# Use only the task logger
logger = get_structured_logger(__name__)
# logger = get_task_logger(__name__)
ops_logger = get_task_logger("ccat_ops_db")
redis_ = get_redis_connection()
_op_config = OperationalConfig()
# The Celery task arg is the TransferOperation id (the sibling Operation row
# that is the source of truth for the transfer lifecycle, ADR-0003), NOT the
# legacy DataTransfer id. The base task (#153) keys mark_in_progress /
# get_retry_count / recovery off this id, so every helper that receives it
# resolves the TransferOperation first.
def _ensure_transfer_operation(
session: Session, data_transfer: models.DataTransfer
) -> models.TransferOperation:
"""Find or create the sibling TransferOperation for a DataTransfer intent.
A TransferOperation is uniquely anchored by
``(data_transfer_package_id, origin_location_id, destination_location_id)``
so a package's primary and secondary transfers each own a distinct row. Kept
idempotent: the find-work loop may re-run before the row is dispatched.
"""
op = (
session.query(models.TransferOperation)
.filter_by(
data_transfer_package_id=data_transfer.data_transfer_package_id,
origin_location_id=data_transfer.origin_location_id,
destination_location_id=data_transfer.destination_location_id,
)
.first()
)
if op is None:
op = models.TransferOperation(
data_transfer_package_id=data_transfer.data_transfer_package_id,
origin_location_id=data_transfer.origin_location_id,
destination_location_id=data_transfer.destination_location_id,
transfer_method=data_transfer.data_transfer_method,
status=models.Status.PENDING,
)
session.add(op)
# Flush so the row has an id before it is dispatched as the Celery task arg.
session.flush()
return op
def _data_transfer_for_operation(
session: Session, transfer_operation: models.TransferOperation
) -> Optional[models.DataTransfer]:
"""Resolve the legacy DataTransfer that carries the rich execution context
(locations, package, logs, process_id) for a TransferOperation.
During the additive transition the BBCP machinery still hangs off the
DataTransfer row; the operation row owns the lifecycle. They are matched on
the same anchor (package + origin + destination)."""
return (
session.query(models.DataTransfer)
.filter_by(
data_transfer_package_id=transfer_operation.data_transfer_package_id,
origin_location_id=transfer_operation.origin_location_id,
destination_location_id=transfer_operation.destination_location_id,
)
.first()
)
[docs]
class DataTransferTask(make_celery_task()):
"""Base class for data transfer tasks.
Keyed on the TransferOperation id (ADR-0003): retry counting falls to the
base uniform ``Operation.retry_count`` reader (the legacy per-row override is
gone), and failure/IN_PROGRESS hooks act on the operation row.
"""
[docs]
def __init__(self):
super().__init__()
self.operation_type = models.OperationKind.TRANSFER.value
self.notification_client = NotificationClient()
[docs]
def mark_in_progress(self, session, transfer_operation_id):
"""Flip the TransferOperation to IN_PROGRESS at task start (the single
seam from the base task, ADR-0003)."""
op = session.query(models.TransferOperation).get(transfer_operation_id)
if op:
op.status = models.Status.IN_PROGRESS
[docs]
def reset_state_on_failure(self, session, transfer_operation_id, exc):
"""Reset the TransferOperation for retry (own status + own retry_count)."""
op = session.query(models.TransferOperation).get(transfer_operation_id)
if op:
op.status = models.Status.PENDING
data_transfer = _data_transfer_for_operation(session, op)
if data_transfer:
for (
raw_data_package
) in data_transfer.data_transfer_package.raw_data_packages:
raw_data_package.state = models.PackageState.TRANSFERRING
op.failure_error_message = None
# error_context mirrors failure_error_message: NULLed on reset (the
# durable trail is kept in OperationFailureEvent). (#117)
op.error_context = None
op.retry_count += 1
logger.info(
"reset transfer for retry",
transfer_operation_id=transfer_operation_id,
retry_count=op.retry_count,
)
[docs]
def mark_permanent_failure(self, session, transfer_operation_id, exc):
"""Mark the TransferOperation permanently FAILED."""
op = session.query(models.TransferOperation).get(transfer_operation_id)
if op:
op.status = models.Status.FAILED
data_transfer = _data_transfer_for_operation(session, op)
if data_transfer:
for (
raw_data_package
) in data_transfer.data_transfer_package.raw_data_packages:
raw_data_package.state = models.PackageState.FAILED
op.failure_error_message = str(exc)
# Cache the latest Tier-1 breadcrumb on the row for the UI (#117).
op.error_context = self._current_error_context
logger.info(
"marked transfer as permanently failed",
transfer_operation_id=transfer_operation_id,
)
[docs]
def get_operation_info(self, args, kwargs):
"""Get additional context for data transfer tasks."""
if not args or len(args) == 0:
return {}
with self.session_scope() as session:
try:
op = session.query(models.TransferOperation).get(args[0])
if op:
data_transfer = _data_transfer_for_operation(session, op)
if data_transfer:
return {
"source_location": data_transfer.origin_location.name,
"destination_location": data_transfer.destination_location.name,
"package_id": str(op.data_transfer_package_id),
}
except Exception as e:
logger.error("failed to get transfer info", error=str(e))
return {}
[docs]
def on_failure(self, exc, task_id, args, kwargs, einfo):
"""Handle task failure with recovery for specific error cases."""
operation_id = self.get_operation_id(args)
if not operation_id:
logger.error("no operation ID found in task arguments")
return
# Pre-cleanup for DestinationFileExistsError before standard failure handling
if isinstance(exc, DestinationFileExistsError):
try:
with self.session_scope() as session:
op = session.query(models.TransferOperation).get(operation_id)
_remove_destination_file(
exc.destination_path,
_data_transfer_for_operation(session, op) if op else None,
)
logger.info(
"removed existing destination file before retry",
operation_id=operation_id,
)
except Exception as e:
logger.error(
"failed to remove destination file",
operation_id=operation_id,
error=str(e),
)
# Call parent's on_failure for retry logic, notifications, and permanent failure handling
super().on_failure(exc, task_id, args, kwargs, einfo)
@app.task(
base=DataTransferTask,
name="ccat:data_transfer:transfer",
bind=True,
)
def transfer_files_bbcp(
self, transfer_operation_id: int, session: Optional[Session] = None
) -> None:
"""Transfer files using BBCP with dynamic queue routing.
The argument is the TransferOperation id (the lifecycle source of truth);
the BBCP execution context is resolved from the matching legacy DataTransfer.
"""
if session is None:
with self.session_scope() as session:
return _transfer_files_bbcp_internal(session, transfer_operation_id)
return _transfer_files_bbcp_internal(session, transfer_operation_id)
@track_metrics(
operation_type="data_transfer",
additional_tags={
"transfer_method": "bbcp",
},
)
def _transfer_files_bbcp_internal(
session: Session, transfer_operation_id: int
) -> None:
"""
Internal function to transfer files using BBCP.
"""
logger = get_structured_logger(__name__)
start_time = datetime.datetime.now()
transfer_operation = session.query(models.TransferOperation).get(
transfer_operation_id
)
if transfer_operation is None:
raise ValueError(
f"Transfer operation not found: {transfer_operation_id}"
)
data_transfer = _data_transfer_for_operation(session, transfer_operation)
if data_transfer is None:
raise ValueError(
"No DataTransfer execution context for transfer operation "
f"{transfer_operation_id}"
)
_log_transfer_start(data_transfer)
transfer_operation.start_time = start_time
data_transfer.start_time = start_time
session.commit()
# Create destination directory if needed (for disk locations)
if isinstance(data_transfer.destination_location, models.DiskDataLocation):
destination_path = os.path.join(
data_transfer.destination_location.path,
data_transfer.data_transfer_package.relative_path,
)
destination_dir = os.path.dirname(destination_path)
logger.debug(
"creating destination directory",
folder=destination_dir,
)
if data_transfer.destination_location.host == "localhost":
create_local_folder(destination_dir)
else:
create_remote_folder(
data_transfer.destination_location.user,
data_transfer.destination_location.host,
destination_dir,
)
source_url, destination_url = _construct_transfer_urls(data_transfer)
result, transfer_metrics = _execute_bbcp_command(
session, data_transfer, source_url, destination_url
)
end_time = datetime.datetime.now()
transfer_operation.end_time = end_time
data_transfer.end_time = end_time
session.commit()
# Verify package arrived intact before marking COMPLETED
if result.returncode == 0:
_verify_destination_checksum(data_transfer)
bbcp_settings = BBCPSettings()
current_settings = bbcp_settings.get_all_settings()
# Send metrics to InfluxDB
metrics = HousekeepingMetrics()
try:
metrics.send_transfer_metrics(
operation="bbcp_transfer",
source_path=source_url,
destination_path=destination_url,
file_size=transfer_metrics["bytes_transferred"],
duration=transfer_metrics["duration"],
success=(result.returncode == 0),
error_message=result.stderr if result.returncode != 0 else None,
additional_fields={
"peak_transfer_rate_mbps": transfer_metrics["peak_transfer_rate_mbps"],
"average_transfer_rate_mbps": transfer_metrics[
"average_transfer_rate_mbps"
],
"number_of_streams": transfer_metrics["number_of_streams"],
"network_errors": transfer_metrics["network_errors"],
"retry_count": transfer_operation.retry_count,
},
additional_tags={
"source_location": data_transfer.origin_location.name,
"destination_location": data_transfer.destination_location.name,
"transfer_id": str(transfer_operation_id),
"transfer_method": "bbcp",
**{
k.lower(): v for k, v in current_settings.items()
}, # Add BBCP settings as tags
},
)
except Exception as e:
logger.error(
"metrics send failed",
error=str(e),
transfer_id=transfer_operation_id,
)
finally:
metrics.close()
# Update status (commits and publishes Redis notification on success)
_update_data_transfer_status(
session,
transfer_operation,
data_transfer,
result,
start_time,
end_time,
transfer_metrics,
)
def _get_data_transfer(session: Session, data_transfer_id: int) -> models.DataTransfer:
"""
Retrieve the data transfer object from the database.
Parameters
----------
session : sqlalchemy.orm.Session
The database session.
data_transfer_id : int
The ID of the data transfer to retrieve.
Returns
-------
models.DataTransfer
The retrieved DataTransfer object.
"""
return session.get(models.DataTransfer, data_transfer_id)
def _log_transfer_start(data_transfer: models.DataTransfer) -> None:
"""Log the start of the data transfer (INFO milestone)."""
logger.info(
"transfer started",
transfer_id=data_transfer.id,
location=data_transfer.origin_location.name,
size_bytes=data_transfer.data_transfer_package.size,
)
def _construct_transfer_urls(data_transfer: models.DataTransfer) -> Tuple[str, str]:
"""
Construct the source and destination URLs for the BBCP command.
Handles different storage types (disk, S3, tape) polymorphically.
Parameters
----------
data_transfer : models.DataTransfer
The DataTransfer object containing the location information.
Returns
-------
Tuple[str, str]
The source and destination URLs.
"""
# Get source and destination paths based on storage type
source_path = _get_location_path(
data_transfer.origin_location, data_transfer.data_transfer_package
)
destination_path = _get_location_path(
data_transfer.destination_location, data_transfer.data_transfer_package
)
# Construct URLs based on storage type
source_url = _construct_url_for_location(data_transfer.origin_location, source_path)
destination_url = _construct_url_for_location(
data_transfer.destination_location, destination_path
)
return source_url, destination_url
def _get_location_path(
data_location: models.DataLocation,
data_transfer_package: models.DataTransferPackage,
) -> str:
"""
Get the full path for a data transfer package at a specific location.
Parameters
----------
data_location : models.DataLocation
The data location.
data_transfer_package : models.DataTransferPackage
The data transfer package.
Returns
-------
str
The full path to the package at this location.
"""
if isinstance(data_location, models.DiskDataLocation):
return safe_join(data_location.path, data_transfer_package.relative_path)
elif isinstance(data_location, models.S3DataLocation):
return f"{data_location.prefix}{data_transfer_package.relative_path}"
elif isinstance(data_location, models.TapeDataLocation):
return safe_join(
data_location.mount_path, data_transfer_package.relative_path
)
else:
raise ValueError(f"Unsupported storage type: {data_location.storage_type}")
def _construct_url_for_location(data_location: models.DataLocation, path: str) -> str:
"""
Construct a URL for a specific location type.
Parameters
----------
data_location : models.DataLocation
The data location.
path : str
The file path.
Returns
-------
str
The constructed URL.
"""
if isinstance(data_location, models.DiskDataLocation):
# Handle localhost connections directly without SSH
if data_location.host == "localhost":
return path
else:
return f"{data_location.user}@{data_location.host}:{path}"
elif isinstance(data_location, models.S3DataLocation):
return f"s3://{data_location.bucket_name}/{path}"
elif isinstance(data_location, models.TapeDataLocation):
# For tape, use local path (tape is mounted locally)
return path
else:
raise ValueError(f"Unsupported storage type: {data_location.storage_type}")
def _remove_destination_file(
destination_url: str, data_transfer: models.DataTransfer
) -> None:
"""Remove destination file if it exists."""
logger = get_structured_logger(__name__)
try:
if isinstance(data_transfer.destination_location, models.DiskDataLocation):
if data_transfer.destination_location.host == "localhost":
if os.path.exists(destination_url):
os.remove(destination_url)
logger.info(
"removed existing destination file",
path=destination_url,
transfer_id=data_transfer.id,
)
else:
# For remote hosts, use SSH to remove the file
ssh_command = [
"ssh", *SSH_OPTS,
f"{data_transfer.destination_location.user}@{data_transfer.destination_location.host}",
f"rm -f {shlex.quote(destination_url)}",
]
result = subprocess.run(
ssh_command, capture_output=True, text=True, check=False
)
if result.returncode == 0:
logger.info(
"removed existing destination file",
path=destination_url,
transfer_id=data_transfer.id,
)
else:
logger.warning(
"failed to remove destination file",
path=destination_url,
error=result.stderr,
transfer_id=data_transfer.id,
)
elif isinstance(data_transfer.destination_location, models.S3DataLocation):
# For S3, use AWS CLI or boto3 to remove file
# This would need to be implemented based on your S3 access method
logger.info(
"s3 file removal not yet implemented",
path=destination_url,
transfer_id=data_transfer.id,
)
else:
logger.warning(
"file removal not implemented for storage type",
storage_type=data_transfer.destination_location.storage_type,
transfer_id=data_transfer.id,
)
except Exception as e:
logger.error(
"error removing destination file",
path=destination_url,
error=str(e),
transfer_id=data_transfer.id,
)
raise
def _execute_bbcp_command(
session: Session,
data_transfer: models.DataTransfer,
source_url: str,
destination_url: str,
):
logger = get_structured_logger(__name__)
# Use cp for localhost-to-localhost transfers (bbcp does not support this)
use_cp = (
isinstance(data_transfer.destination_location, models.DiskDataLocation)
and data_transfer.destination_location.host == "localhost"
)
if use_cp:
return _execute_cp_command(session, data_transfer, source_url, destination_url)
bbcp_command = make_bbcp_command(source_url, destination_url)
logger.debug("executing BBCP command", command=bbcp_command)
log_handler = BBCPLogHandler()
start_time = time.time()
# check if the remote destination file already exists (for disk locations)
if (
isinstance(data_transfer.destination_location, models.DiskDataLocation)
and data_transfer.destination_location.host != "localhost"
):
ssh_command = [
"ssh", *SSH_OPTS,
f"{data_transfer.destination_location.user}@{data_transfer.destination_location.host}",
f"ls {shlex.quote(destination_url)}",
]
result = subprocess.run(
ssh_command, capture_output=True, text=True, check=False
)
if result.returncode == 0:
logger.info(
"destination file already exists",
path=destination_url,
transfer_id=data_transfer.id,
)
# remove the file from the destination
ssh_command = [
"ssh", *SSH_OPTS,
f"{data_transfer.destination_location.user}@{data_transfer.destination_location.host}",
f"rm -f {shlex.quote(destination_url)}",
]
subprocess.run(ssh_command, capture_output=True, text=True, check=False)
# Separate try block just for subprocess execution
try:
result = subprocess.run(
bbcp_command,
capture_output=True,
text=True,
check=False,
)
except subprocess.SubprocessError as e:
raise BBCPError(
message="Failed to execute BBCP command",
returncode=-1,
stderr=str(e),
transfer_id=data_transfer.id,
)
if result.returncode != 0:
if (
"Connection refused" in result.stderr
or "Connection timed out" in result.stderr
):
raise NetworkError(
result.stderr,
host=(
data_transfer.destination_location.host
if isinstance(
data_transfer.destination_location, models.DiskDataLocation
)
else "unknown"
),
transfer_id=data_transfer.id,
)
elif result.returncode == -11:
raise SegmentationFaultError(
message="Segmentation Fault",
returncode=result.returncode,
stderr=result.stderr,
transfer_id=data_transfer.id,
)
elif "already exists" in result.stderr:
# Handle destination file exists error
raise DestinationFileExistsError(
message=result.stderr,
returncode=result.returncode,
stderr=result.stderr,
transfer_id=data_transfer.id,
destination_path=destination_url,
)
else:
raise BBCPError(
message=result.stderr,
returncode=result.returncode,
stderr=result.stderr,
transfer_id=data_transfer.id,
)
end_time = time.time()
duration = end_time - start_time
logger.debug(
"bbcp_complete",
return_code=result.returncode,
transfer_id=data_transfer.id,
duration=duration,
)
# Store command output
log_handler.store_bbcp_output(
session=session,
data_transfer=data_transfer,
stdout=result.stdout,
stderr=result.stderr,
success=(result.returncode == 0),
)
# Parse metrics
metrics = parse_bbcp_output(result.stdout, result.stderr, duration)
return result, metrics
def _execute_cp_command(
session: Session,
data_transfer: models.DataTransfer,
source_path: str,
destination_path: str,
):
"""Execute cp command for localhost-to-localhost transfers."""
logger = get_structured_logger(__name__)
log_handler = BBCPLogHandler()
start_time = time.time()
try:
# Guard the source read so a missing source surfaces as a diagnosed
# SourceMissingError instead of a bare FileNotFoundError (#118). The
# diagnoser explains *why* the DataTransferPackage source is gone.
origin_location = data_transfer.origin_location
ensure_readable(
source_path,
side="source",
step="pre_transfer_stat",
host=getattr(origin_location, "host", None),
diagnose=lambda: diagnose_source_missing(
session,
models.DataTransferPackagePhysicalCopy,
fk_attr="data_transfer_package_id",
fk_value=data_transfer.data_transfer_package_id,
location=origin_location,
),
)
# Get file size before transfer
file_size = os.path.getsize(source_path)
cp_command = ["cp", "-p", source_path, destination_path]
logger.debug(
"executing cp command",
command=cp_command,
size_bytes=file_size,
)
# Execute cp command
result = subprocess.run(
cp_command,
capture_output=True,
text=True,
check=False,
)
end_time = time.time()
duration = end_time - start_time
if result.returncode != 0:
raise BBCPError(
message="Failed to execute cp command",
returncode=result.returncode,
stderr=result.stderr,
transfer_id=data_transfer.id,
)
# Calculate metrics similar to bbcp
transfer_rate = (file_size / duration) / (1024 * 1024) # MB/s
metrics = {
"bytes_transferred": file_size,
"duration": duration,
"peak_transfer_rate_mbps": transfer_rate,
"average_transfer_rate_mbps": transfer_rate,
"number_of_streams": 1,
"network_errors": 0,
}
# Store command output
log_handler.store_bbcp_output(
session=session,
data_transfer=data_transfer,
stdout=f"Copied {file_size} bytes in {duration:.2f} seconds ({transfer_rate:.2f} MB/s)",
stderr=result.stderr,
success=(result.returncode == 0),
)
logger.debug(
"cp_complete",
return_code=result.returncode,
transfer_id=data_transfer.id,
duration=duration,
)
return result, metrics
except subprocess.SubprocessError as e:
raise BBCPError(
message="Failed to execute cp command",
returncode=-1,
stderr=str(e),
transfer_id=data_transfer.id,
)
def _verify_destination_checksum(data_transfer: models.DataTransfer) -> None:
"""
Verify the package checksum at the destination after a successful transfer.
Compares the xxHash64 checksum stored in the DB against the arrived file.
Raises ArchiveCorruptionError on mismatch so the task system can retry.
Parameters
----------
data_transfer : models.DataTransfer
The completed data transfer.
Raises
------
ArchiveCorruptionError
If the destination checksum does not match the stored value.
"""
from .utils import calculate_checksum
logger = get_structured_logger(__name__)
expected = data_transfer.data_transfer_package.checksum
if not expected:
logger.warning(
"skipping post-transfer checksum: no stored checksum",
transfer_id=data_transfer.id,
)
return
destination_path = _get_location_path(
data_transfer.destination_location, data_transfer.data_transfer_package
)
if not isinstance(data_transfer.destination_location, models.DiskDataLocation):
logger.warning(
"post-transfer checksum not supported for non-disk destination",
transfer_id=data_transfer.id,
storage_type=str(data_transfer.destination_location.storage_type),
)
return
if data_transfer.destination_location.host == "localhost":
computed = calculate_checksum(destination_path)
else:
host = data_transfer.destination_location.host
user = data_transfer.destination_location.user
ssh_command = [
"ssh", *SSH_OPTS,
f"{user}@{host}",
f"set -o pipefail; xxh64sum {shlex.quote(destination_path)} | cut -d' ' -f1",
]
result = subprocess.run(ssh_command, capture_output=True, text=True, check=False)
if result.returncode != 0:
raise ArchiveCorruptionError(
f"Failed to compute remote checksum: {result.stderr.strip()}",
archive_path=destination_path,
transfer_id=data_transfer.id,
)
computed = result.stdout.strip()
if not computed:
raise ArchiveCorruptionError(
f"Remote checksum command returned empty output (stderr: {result.stderr.strip()})",
archive_path=destination_path,
transfer_id=data_transfer.id,
)
if computed != expected:
raise ArchiveCorruptionError(
f"Post-transfer checksum mismatch: expected {expected}, got {computed}",
archive_path=destination_path,
transfer_id=data_transfer.id,
)
logger.info(
"post-transfer checksum verified",
transfer_id=data_transfer.id,
checksum=computed,
)
def _update_data_transfer_status(
session: Session,
transfer_operation: models.TransferOperation,
data_transfer: models.DataTransfer,
result: subprocess.CompletedProcess,
start_time: datetime.datetime,
end_time: datetime.datetime,
transfer_metrics: Dict[str, Any],
) -> None:
"""Update the transfer operation (and the mirrored legacy row) in the DB."""
logger = get_structured_logger(__name__)
duration_s = (end_time - start_time).total_seconds()
# Computed once so it is in scope for both the success (physical copy) and
# failure (DestinationFileExistsError) branches.
destination_path = _get_location_path(
data_transfer.destination_location, data_transfer.data_transfer_package
)
if result.returncode == 0:
logger.info(
"transfer succeeded",
transfer_id=data_transfer.id,
location=data_transfer.origin_location.name,
size_bytes=transfer_metrics.get("bytes_transferred"),
duration_s=duration_s,
)
transfer_operation.start_time = start_time
transfer_operation.end_time = end_time
transfer_operation.status = models.Status.COMPLETED
# Stamp the transfer-phase timestamps on the legacy row (kept columns);
# lifecycle status lives only on the TransferOperation now (#95).
data_transfer.start_time = start_time
data_transfer.end_time = end_time
# Produce the destination copy; transfer's output in the copy-anchored
# lineage (ADR-0003). The sibling unpack operation consumes it.
physical_copy = models.DataTransferPackagePhysicalCopy(
data_transfer_package=data_transfer.data_transfer_package,
data_location=data_transfer.destination_location,
status=models.PhysicalCopyStatus.PRESENT,
checksum=data_transfer.data_transfer_package.checksum,
)
session.add(physical_copy)
transfer_operation.produced_copies.append(physical_copy)
# Consume the source copy at the origin, if one is recorded.
source_copy = (
session.query(models.DataTransferPackagePhysicalCopy)
.filter_by(
data_transfer_package_id=data_transfer.data_transfer_package_id,
data_location_id=data_transfer.origin_location_id,
status=models.PhysicalCopyStatus.PRESENT,
)
.first()
)
if source_copy is not None:
transfer_operation.consumed_copies.append(source_copy)
logger.info("storing physical copy record", transfer_id=data_transfer.id)
else:
logger.error(
"transfer failed",
transfer_id=data_transfer.id,
location=data_transfer.origin_location.name,
size_bytes=transfer_metrics.get("bytes_transferred"),
duration_s=duration_s,
return_code=result.returncode,
error=result.stderr.strip(),
)
# Instead of directly updating status, raise an exception to trigger retry handling
if (
"Connection refused" in result.stderr
or "Connection timed out" in result.stderr
):
raise NetworkError(
result.stderr,
host=(
data_transfer.destination_location.host
if isinstance(
data_transfer.destination_location, models.DiskDataLocation
)
else "unknown"
),
transfer_id=data_transfer.id,
)
elif result.returncode == -11:
raise SegmentationFaultError(
message="Segmentation Fault",
returncode=result.returncode,
stderr=result.stderr,
transfer_id=data_transfer.id,
)
elif "already exists" in result.stderr:
raise DestinationFileExistsError(
message=result.stderr,
returncode=result.returncode,
stderr=result.stderr,
transfer_id=data_transfer.id,
destination_path=destination_path,
)
else:
raise BBCPError(
message=result.stderr,
returncode=result.returncode,
stderr=result.stderr,
transfer_id=data_transfer.id,
)
session.commit()
redis_.publish(
"transfer:overview",
json.dumps({"type": "transfer_completed", "data": data_transfer.id}),
)
def _get_pending_transfers(session: Session) -> List[models.DataTransfer]:
"""
Retrieve all pending data transfers from the database.
Parameters
----------
session : sqlalchemy.orm.Session
The database session.
Returns
-------
List[models.DataTransfer]
A list of pending DataTransfer objects.
Notes
-----
The transfer lifecycle moved onto the sibling TransferOperation (ADR-0003);
DataTransfer.status was dropped in ops-db#95. A DataTransfer is pending when
its sibling TransferOperation (anchored on package + origin + destination) is
PENDING, or does not exist yet AND the transfer has not already finished —
the find-work loop creates the op via _ensure_transfer_operation and
dispatches it. Once dispatched the op leaves PENDING (SCHEDULED/IN_PROGRESS/
...), so the row drops out of this query.
The ``end_time IS NULL`` guard on the op-less arm is required for brownfield
safety: rows created before the operation model have no TransferOperation but
did complete (end_time set), and without the guard they would be resurrected
and re-dispatched as bbcp jobs whose source files are long gone. end_time is
the only surviving completion signal once the legacy status column is dropped.
"""
pending_transfers = (
session.query(models.DataTransfer)
.outerjoin(
models.TransferOperation,
(
models.TransferOperation.data_transfer_package_id
== models.DataTransfer.data_transfer_package_id
)
& (
models.TransferOperation.origin_location_id
== models.DataTransfer.origin_location_id
)
& (
models.TransferOperation.destination_location_id
== models.DataTransfer.destination_location_id
),
)
.filter(
or_(
and_(
models.TransferOperation.id.is_(None),
models.DataTransfer.end_time.is_(None),
),
models.TransferOperation.status == models.Status.PENDING,
)
)
.all()
)
return pending_transfers
def _filter_supported_transfers(
transfers: List[models.DataTransfer],
) -> Tuple[List[models.DataTransfer], List[str]]:
"""
Filter transfers based on supported transfer methods.
Parameters
----------
transfers : List[models.DataTransfer]
A list of DataTransfer objects to filter.
Returns
-------
Tuple[List[models.DataTransfer], List[str]]
A tuple containing a list of supported transfers and a list of unsupported transfer methods.
"""
supported = []
unsupported_methods = set()
for transfer in transfers:
if (
transfer.data_transfer_method
in ccat_data_transfer_settings.SUPPORTED_DATA_TRANSFER_METHODS
):
supported.append(transfer)
else:
unsupported_methods.add(transfer.data_transfer_method)
return supported, list(unsupported_methods)
def _process_transfer(transfer: models.DataTransfer, session: Session) -> None:
"""Process a single data transfer.
Ensures the sibling TransferOperation exists and dispatches the Celery task
with the OPERATION id — the operation row is the lifecycle source of truth
(ADR-0003). The legacy DataTransfer status is mirrored for backward-compat.
"""
logger = get_structured_logger(__name__)
transfer_operation = _ensure_transfer_operation(session, transfer)
# Check if we can create new data based on buffer state
if not buffer_manager.can_create_data():
logger.warning(
"buffer in emergency state, postponing transfer",
transfer_id=transfer.id,
file=transfer.data_transfer_package.file_name,
)
transfer_operation.status = models.Status.PENDING
session.commit()
redis_.publish(
"transfer:overview",
json.dumps({"type": "transfer_pending", "data": transfer.id}),
)
return
if transfer.data_transfer_method == "bbcp":
# Use dynamic queue routing based on origin location
queue_name = route_task_by_location(
OperationType.DATA_TRANSFER, transfer.origin_location
)
task_args = {
"args": (transfer_operation.id,),
"queue": queue_name,
}
# Get max parallel transfers based on buffer state
max_transfers = buffer_manager.get_max_parallel_transfers()
# Apply rate limiting based on buffer state
data_transfer_workers = _op_config.get(session, "DATA_TRANSFER_WORKERS")
if max_transfers < data_transfer_workers:
logger.info(
"reducing parallel transfers due to buffer state",
max_transfers=max_transfers,
normal_workers=data_transfer_workers,
)
# Use a rate-limited queue
task_args["queue"] = f"{queue_name}-limited"
# Apply the task using the unified transfer function
transfer_files_bbcp.apply_async(**task_args)
transfer_operation.status = models.Status.SCHEDULED
session.commit()
redis_.publish(
"transfer:overview",
json.dumps({"type": "transfer_scheduled", "data": transfer.id}),
)
logger.debug(
"transfer scheduled",
transfer_id=transfer.id,
transfer_operation_id=transfer_operation.id,
file=transfer.data_transfer_package.file_name,
source=transfer.origin_location.name,
queue=queue_name,
)
[docs]
def transfer_transfer_packages(verbose: bool = False, session: Session = None) -> None:
"""
Find not yet transferred data transfer packages and schedule their transfer.
Parameters
----------
verbose : bool, optional
If True, sets the logging level to DEBUG. Default is False.
Returns
-------
None
Notes
-----
- Updates the logging level if verbose is True.
- Retrieves pending data transfers from the database.
- Schedules Celery tasks for file transfers.
- Updates data transfer statuses in the database.
- Logs information about the transfer process.
- Handles database errors and unexpected exceptions.
"""
db = DatabaseConnection()
should_close_session = False
if session is None:
db = DatabaseConnection()
session, _ = db.get_connection()
should_close_session = True
try:
pending_transfers = _get_pending_transfers(session)
supported_transfers, unsupported_methods = _filter_supported_transfers(
pending_transfers
)
if unsupported_methods:
logger.error("unsupported transfer methods found", methods=",".join(unsupported_methods))
if len(pending_transfers) > 0:
logger.debug(
"pending transfers found",
count=len(pending_transfers),
first_id=pending_transfers[0].id,
)
else:
logger.debug("no pending transfers")
for transfer in supported_transfers:
_process_transfer(transfer, session)
except Exception as e:
logger.error("service loop error", error=str(e))
finally:
if should_close_session:
session.close()
