Implement resumable file transfers

- Switched from Redis lists to Redis Streams for chunk tracking
- Added upload/download progress persistence and resumption
- Implemented WebSocket /resume/{uid} endpoint for upload resumption
- Added HTTP Range header support for partial downloads (206 responses)
- Created resumption handler for managing transfer state
- Updated JavaScript client to save/restore upload progress via localStorage
- Added connection management for handling disconnections gracefully
- Transfers now wait for peer reconnection within timeout window
- All transfers are resumable by default (no configuration needed)

Key components:
- lib/store.py: Refactored to use Redis Streams with progress tracking
- lib/stream_store.py: New module for stream operations (deprecated)
- lib/transfer.py: Enhanced with resume_from and start_byte parameters
- lib/resume.py: Handles resumption logic and state validation
- lib/range_utils.py: Parses and validates HTTP Range headers
- lib/connection.py: Manages peer states and reconnection windows
- views/websockets.py: Added /resume endpoint for upload resumption
- views/http.py: Enhanced with Range/partial content support
- static/js/file-transfer.js: Added localStorage progress persistence

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <noreply@anthropic.com>

Author: codeSamuraii

lib/connection.py

@@ -0,0 +1,125 @@
+import anyio
+from typing import Optional, Dict, Any
+from datetime import datetime
+from lib.logging import HasLogging, get_logger
+from lib.store import Store
+
+logger = get_logger('connection')
+
+
+class ConnectionManager(metaclass=HasLogging, name_from='transfer_id'):
+    """Manages connection states and reconnection logic for resumable transfers."""
+
+    RECONNECT_WINDOW = 60.0  # seconds
+    KEEPALIVE_INTERVAL = 10.0  # seconds
+
+    def __init__(self, transfer_id: str, store: Store):
+        self.transfer_id = transfer_id
+        self.store = store
+        self.peer_states: Dict[str, Any] = {}
+
+    async def register_peer(self, peer_type: str, connection_info: Dict[str, Any]) -> None:
+        """Register a peer connection."""
+        state_data = {
+            'status': 'connected',
+            'timestamp': datetime.now().isoformat(),
+            **connection_info
+        }
+        await self.store.set_peer_state(peer_type, 'connected')
+        self.peer_states[peer_type] = state_data
+        self.info(f"Registered {peer_type} connection")
+
+    async def handle_disconnect(self, peer_type: str) -> bool:
+        """Handle peer disconnection and return whether to wait for reconnection."""
+        await self.store.set_peer_state(peer_type, 'disconnected')
+
+        other_peer = 'receiver' if peer_type == 'sender' else 'sender'
+        other_state = await self.store.get_peer_state(other_peer)
+
+        if other_state in ['connected', 'uploading', 'downloading']:
+            self.info(f"{peer_type} disconnected, keeping {other_peer} connected")
+            return True  # Wait for reconnection
+
+        self.warning(f"Both peers disconnected, may abandon transfer")
+        return False
+
+    async def wait_for_reconnection(self, peer_type: str, timeout: Optional[float] = None) -> bool:
+        """Wait for a peer to reconnect within timeout."""
+        timeout = timeout or self.RECONNECT_WINDOW
+        self.info(f"Waiting up to {timeout}s for {peer_type} to reconnect...")
+
+        try:
+            with anyio.fail_after(timeout):
+                while True:
+                    state = await self.store.get_peer_state(peer_type)
+                    if state in ['connected', 'uploading', 'downloading', 'resuming']:
+                        self.info(f"{peer_type} reconnected successfully")
+                        return True
+                    await anyio.sleep(1.0)
+        except TimeoutError:
+            self.warning(f"{peer_type} did not reconnect within {timeout}s")
+            return False
+
+    async def handle_reconnection(self, peer_type: str) -> Dict[str, Any]:
+        """Handle peer reconnection and return resume info."""
+        await self.store.set_peer_state(peer_type, 'resuming')
+
+        # Check if other peer is waiting
+        other_peer = 'receiver' if peer_type == 'sender' else 'sender'
+        other_state = await self.store.get_peer_state(other_peer)
+
+        resume_info = {}
+        if peer_type == 'sender':
+            progress = await self.store.get_upload_progress()
+            if progress:
+                resume_info['bytes_uploaded'] = progress['bytes_uploaded']
+                resume_info['last_chunk_id'] = progress['last_chunk_id']
+        else:
+            progress = await self.store.get_download_progress()
+            if progress:
+                resume_info['bytes_downloaded'] = progress['bytes_downloaded']
+                resume_info['last_read_id'] = progress['last_read_id']
+
+        resume_info['other_peer_state'] = other_state
+        resume_info['can_resume'] = True
+
+        self.info(f"{peer_type} reconnection handled, resume info: {resume_info}")
+        return resume_info
+
+    async def keepalive_loop(self, peer_type: str) -> None:
+        """Send keepalive signals to maintain connection state."""
+        while True:
+            try:
+                await anyio.sleep(self.KEEPALIVE_INTERVAL)
+                state = await self.store.get_peer_state(peer_type)
+                if state not in ['connected', 'uploading', 'downloading']:
+                    break
+                # Update timestamp to show peer is still alive
+                await self.store.set_peer_state(peer_type, state)
+            except Exception as e:
+                self.error(f"Keepalive error for {peer_type}: {e}")
+                break
+
+    async def check_peer_health(self, peer_type: str) -> bool:
+        """Check if a peer connection is healthy."""
+        state = await self.store.get_peer_state(peer_type)
+        return state in ['connected', 'uploading', 'downloading', 'resuming']
+
+    async def coordinate_resume(self) -> bool:
+        """Coordinate resume between both peers."""
+        sender_state = await self.store.get_peer_state('sender')
+        receiver_state = await self.store.get_peer_state('receiver')
+
+        if sender_state in ['resuming', 'connected'] and receiver_state in ['resuming', 'connected']:
+            self.info("Both peers ready to resume transfer")
+            await self.store.set_event('resume_transfer')
+            return True
+
+        self.debug(f"Cannot resume yet - sender: {sender_state}, receiver: {receiver_state}")
+        return False
+
+    async def cleanup_on_error(self) -> None:
+        """Clean up connection states on error."""
+        await self.store.set_peer_state('sender', 'error')
+        await self.store.set_peer_state('receiver', 'error')
+        self.warning("Connection states cleaned up due to error")

lib/range_utils.py

@@ -0,0 +1,112 @@
+from typing import Optional, Tuple, List
+from dataclasses import dataclass
+
+
+@dataclass
+class RangeRequest:
+    """Represents a parsed HTTP Range request."""
+    start: int
+    end: Optional[int]
+    total_size: Optional[int] = None
+
+    @property
+    def length(self) -> Optional[int]:
+        """Calculate the length of the requested range."""
+        if self.end is not None:
+            return self.end - self.start + 1
+        elif self.total_size is not None:
+            return self.total_size - self.start
+        return None
+
+    def to_content_range(self, total_size: int) -> str:
+        """Generate Content-Range header value."""
+        end = self.end if self.end is not None else total_size - 1
+        return f"bytes {self.start}-{end}/{total_size}"
+
+
+class RangeParser:
+    """Parses and validates HTTP Range headers."""
+
+    @staticmethod
+    def parse_range_header(range_header: Optional[str], file_size: int) -> Optional[RangeRequest]:
+        """Parse Range header and return RangeRequest object."""
+        if not range_header or not range_header.startswith('bytes='):
+            return None
+
+        try:
+            range_spec = range_header[6:]  # Remove 'bytes='
+
+            if ',' in range_spec:
+                # Multiple ranges not supported for now
+                return None
+
+            if '-' not in range_spec:
+                return None
+
+            parts = range_spec.split('-', 1)
+            start_str, end_str = parts[0], parts[1]
+
+            # Handle suffix-length syntax (e.g., "-500" for last 500 bytes)
+            if not start_str and end_str:
+                suffix_length = int(end_str)
+                start = max(0, file_size - suffix_length)
+                end = file_size - 1
+                return RangeRequest(start=start, end=end, total_size=file_size)
+
+            # Handle normal range
+            start = int(start_str) if start_str else 0
+            end = int(end_str) if end_str else file_size - 1
+
+            # Validate range
+            if start < 0 or start >= file_size:
+                return None
+            if end >= file_size:
+                end = file_size - 1
+            if start > end:
+                return None
+
+            return RangeRequest(start=start, end=end, total_size=file_size)
+
+        except (ValueError, IndexError):
+            return None
+
+    @staticmethod
+    def validate_if_range(if_range_header: Optional[str], etag: Optional[str]) -> bool:
+        """Validate If-Range header against ETag."""
+        if not if_range_header or not etag:
+            return True  # No validation needed
+        return if_range_header == etag
+
+    @staticmethod
+    def calculate_chunk_range(chunk_index: int, chunk_size: int, byte_offset: int) -> Tuple[int, int]:
+        """Calculate byte range for a specific chunk."""
+        chunk_start = chunk_index * chunk_size
+        chunk_end = chunk_start + chunk_size - 1
+
+        if chunk_start < byte_offset:
+            # Partial chunk at the beginning
+            return byte_offset, chunk_end
+        return chunk_start, chunk_end
+
+    @staticmethod
+    def is_partial_content(range_request: Optional[RangeRequest]) -> bool:
+        """Check if this is a partial content request."""
+        return range_request is not None and (
+            range_request.start > 0 or
+            (range_request.end is not None and range_request.total_size is not None and
+             range_request.end < range_request.total_size - 1)
+        )
+
+    @staticmethod
+    def create_content_headers(range_request: RangeRequest, file_type: str) -> dict:
+        """Create response headers for partial content."""
+        headers = {
+            'Content-Type': file_type,
+            'Accept-Ranges': 'bytes',
+            'Content-Range': range_request.to_content_range(range_request.total_size)
+        }
+
+        if range_request.length:
+            headers['Content-Length'] = str(range_request.length)
+
+        return headers

lib/resume.py

@@ -0,0 +1,118 @@
+from typing import Optional, Tuple
+from lib.store import Store
+from lib.metadata import FileMetadata
+from lib.logging import HasLogging, get_logger
+
+logger = get_logger('resume')
+
+
+class ResumptionHandler(metaclass=HasLogging, name_from='transfer_id'):
+    """Handles transfer resumption logic."""
+
+    def __init__(self, transfer_id: str, store: Store):
+        self.transfer_id = transfer_id
+        self.store = store
+
+    async def can_resume_upload(self) -> bool:
+        """Check if an upload can be resumed."""
+        progress = await self.store.get_upload_progress()
+        if not progress:
+            return False
+
+        sender_state = await self.store.get_peer_state('sender')
+        return sender_state in ['paused', 'disconnected', 'incomplete']
+
+    async def can_resume_download(self) -> bool:
+        """Check if a download can be resumed."""
+        progress = await self.store.get_download_progress()
+        if not progress:
+            return False
+
+        receiver_state = await self.store.get_peer_state('receiver')
+        return receiver_state in ['paused', 'disconnected', 'sender_disconnected']
+
+    async def get_upload_resume_info(self) -> Tuple[int, str]:
+        """Get upload resume position and last chunk ID."""
+        progress = await self.store.get_upload_progress()
+        if progress:
+            return progress['bytes_uploaded'], progress['last_chunk_id']
+        return 0, '0'
+
+    async def get_download_resume_info(self) -> Tuple[int, str]:
+        """Get download resume position and last read ID."""
+        progress = await self.store.get_download_progress()
+        if progress:
+            return progress['bytes_downloaded'], progress['last_read_id']
+        return 0, '0'
+
+    async def prepare_upload_resume(self) -> dict:
+        """Prepare upload for resumption and return resume info."""
+        bytes_uploaded, last_chunk_id = await self.get_upload_resume_info()
+
+        await self.store.set_peer_state('sender', 'resuming')
+
+        return {
+            'resume_from': bytes_uploaded,
+            'last_chunk_id': last_chunk_id,
+            'can_resume': True
+        }
+
+    async def prepare_download_resume(self, range_header: Optional[str] = None) -> dict:
+        """Prepare download for resumption and return resume info."""
+        if range_header:
+            start_byte = self._parse_range_header(range_header)
+        else:
+            bytes_downloaded, _ = await self.get_download_resume_info()
+            start_byte = bytes_downloaded
+
+        await self.store.set_peer_state('receiver', 'resuming')
+
+        return {
+            'start_byte': start_byte,
+            'can_resume': True,
+            'total_size': None  # Will be filled from metadata
+        }
+
+    def _parse_range_header(self, range_header: str) -> int:
+        """Parse Range header to get start byte position."""
+        if not range_header or not range_header.startswith('bytes='):
+            return 0
+
+        try:
+            range_spec = range_header[6:]  # Remove 'bytes='
+            if '-' in range_spec:
+                start, end = range_spec.split('-', 1)
+                return int(start) if start else 0
+        except (ValueError, IndexError):
+            pass
+
+        return 0
+
+    async def validate_resume_request(self, file: FileMetadata) -> bool:
+        """Validate that resume request matches original transfer."""
+        stored_metadata = await self.store.get_metadata()
+        if not stored_metadata:
+            return False
+
+        try:
+            stored_file = FileMetadata.from_json(stored_metadata)
+            return (stored_file.name == file.name and
+                    stored_file.size == file.size and
+                    stored_file.type == file.type)
+        except Exception:
+            return False
+
+    async def handle_peer_reconnection(self, peer_type: str) -> None:
+        """Handle when a peer reconnects."""
+        other_peer = 'receiver' if peer_type == 'sender' else 'sender'
+        other_state = await self.store.get_peer_state(other_peer)
+
+        if other_state == 'waiting':
+            self.info(f"Both peers reconnected, resuming transfer")
+            await self.store.set_event('resume_transfer')
+
+    async def cleanup_stale_transfers(self, max_age_seconds: int = 3600) -> None:
+        """Clean up stale transfer data older than max_age."""
+        # This would be called periodically to clean up abandoned transfers
+        # Implementation depends on Redis TTL or timestamp tracking
+        pass