Update app.py

app.py

@@ -4,6 +4,7 @@ from fastapi.templating import Jinja2Templates
 from fastapi.staticfiles import StaticFiles
 from fastapi.middleware.cors import CORSMiddleware
 import os
+import re
 import subprocess
 import threading
 import time
@@ -66,6 +67,7 @@ migration_state = {
     "last_activity": time.time(),
     "log": [],
     "process": None,
+    "chunk_info_internal": None,  # NEW: internal TSDB chunk details (not returned to client)
     "progress": {
         "current_table": None,
         "tables_completed": 0,
@@ -73,10 +75,16 @@ migration_state = {
         "current_size_mb": 0,
         "growth_rate_mb_per_sec": 0,
         "estimated_time_remaining": None,
-        "percent_complete": 0
+        "percent_complete": 0,
+        "total_expected_bytes": 0,     # NEW: expected total bytes from chunk map
+        "bytes_completed": 0,          # NEW: bytes completed (chunk-based)
+        "chunks_completed": 0,         # NEW: chunks completed (count)
+        "chunks_total": 0,             # NEW: total chunks (from chunk map)
+        "counted_chunk_names": []      # NEW: to prevent double counting (internal)
     }
 }
 
+
 # Lock for updating global state
 migration_lock = threading.Lock()
 
@@ -115,6 +123,69 @@ def get_file_size_mb(file_path: str) -> float:
     except Exception:
         return 0
 
+def build_timescaledb_chunk_info(conn):
+    """Build a comprehensive map of all TimescaleDB chunks and their compressed counterparts."""
+    info = {
+        "generated_at": datetime.datetime.utcnow().isoformat() + 'Z',
+        "summary": {
+            "chunks_count": 0,
+            "hypertables_count": 0,
+            "total_effective_size_bytes": 0
+        },
+        "name_to_entry": {},  # map of schema.table -> entry
+        "entries": []         # list of chunk relationship entries
+    }
+    try:
+        with conn.cursor() as cur:
+            cur.execute("""
+                SELECT
+                    ht.id AS hypertable_id,
+                    ht.schema_name || '.' || ht.table_name AS hypertable,
+                    orig.schema_name || '.' || orig.table_name AS original_chunk,
+                    CASE WHEN comp.id IS NOT NULL THEN comp.schema_name || '.' || comp.table_name END AS compressed_chunk,
+                    pg_total_relation_size(orig.schema_name || '.' || orig.table_name) AS original_size_bytes,
+                    CASE WHEN comp.id IS NOT NULL THEN pg_total_relation_size(comp.schema_name || '.' || comp.table_name) END AS compressed_size_bytes,
+                    (orig.compressed_chunk_id IS NOT NULL) AS is_compressed,
+                    (ht.table_name LIKE '_materialized_hypertable_%') AS is_cagg
+                FROM _timescaledb_catalog.chunk AS orig
+                JOIN _timescaledb_catalog.hypertable AS ht ON ht.id = orig.hypertable_id
+                LEFT JOIN _timescaledb_catalog.chunk AS comp ON orig.compressed_chunk_id = comp.id
+            """)
+            rows = cur.fetchall()
+            hyps = set()
+            total_effective = 0
+            for (
+                hypertable_id, hypertable, original_chunk, compressed_chunk,
+                original_size, compressed_size, is_compressed, is_cagg
+            ) in rows:
+                effective_size = compressed_size if compressed_size is not None else original_size
+                entry = {
+                    "hypertable_id": hypertable_id,
+                    "hypertable": hypertable,
+                    "original_chunk": original_chunk,
+                    "compressed_chunk": compressed_chunk,
+                    "original_size_bytes": int(original_size) if original_size is not None else 0,
+                    "compressed_size_bytes": int(compressed_size) if compressed_size is not None else None,
+                    "effective_size_bytes": int(effective_size) if effective_size is not None else 0,
+                    "compression_status": "COMPRESSED" if is_compressed else "UNCOMPRESSED",
+                    "is_cagg": bool(is_cagg)
+                }
+                info["entries"].append(entry)
+                hyps.add(hypertable)
+                total_effective += entry["effective_size_bytes"]
+                # Map both names to same entry for fast lookup
+                info["name_to_entry"][original_chunk] = entry
+                if compressed_chunk:
+                    info["name_to_entry"][compressed_chunk] = entry
+
+            info["summary"]["chunks_count"] = len(info["entries"])
+            info["summary"]["hypertables_count"] = len(hyps)
+            info["summary"]["total_effective_size_bytes"] = total_effective
+    except Exception as e:
+        logger.warning(f"Could not build TimescaleDB chunk map: {e}")
+    return info
+
+
 def monitor_dump_size():
     """Monitor the dump file size and update state"""
     while migration_state["running"] and migration_state["operation"] == "dump":
@@ -140,48 +211,35 @@ def monitor_dump_size():
         time.sleep(1)  # Update every second
 
 def run_dump(source_conn: str, file_path: str, options: dict):
-    """Run pg_dump in a background thread"""
+    """Run pg_dump in a background thread with chunk-based progress tracking."""
     try:
-        # Clear any existing file
         if os.path.exists(file_path):
             os.remove(file_path)
 
-        # Set environment variables for connection
         env = os.environ.copy()
-
-        # Build pg_dump command
-        format_flag = "-F" + options.get("format", "c")  # Default to custom format
+        format_flag = "-F" + options.get("format", "c")
         cmd = ["pg_dump", source_conn, format_flag, "-v", "-f", file_path]
 
-        # Add schema if specified
         if options.get("schema"):
             cmd.extend(["-n", options["schema"]])
 
-        # Add compression level if specified
         if options.get("compression") and options["compression"] != "default":
             cmd.extend(["-Z", options["compression"]])
 
         log_message(f"Starting database dump to {file_path}", "info", " ".join(cmd))
 
-        # Start monitoring thread for file size
         monitor_thread = threading.Thread(target=monitor_dump_size, daemon=True)
         monitor_thread.start()
 
-        # Start the dump process
         with migration_lock:
             migration_state["start_time"] = time.time()
             migration_state["running"] = True
             migration_state["operation"] = "dump"
             migration_state["dump_file"] = file_path
             migration_state["dump_completed"] = False
-            migration_state["previous_size"] = 0 # Reset previous size
+            migration_state["previous_size"] = 0
 
-        # Use preexec_fn=os.setsid to create a new process group
-        # This is necessary for the killpg logic in stop_current_process
-        # Note: preexec_fn is Unix-specific. This won't work directly on Windows.
-        preexec_fn_to_use = None
-        if hasattr(os, 'setsid'):
-            preexec_fn_to_use = os.setsid
+        preexec_fn_to_use = os.setsid if hasattr(os, 'setsid') else None
 
         process = subprocess.Popen(
             cmd,
@@ -189,49 +247,89 @@ def run_dump(source_conn: str, file_path: str, options: dict):
             stderr=subprocess.PIPE,
             env=env,
             text=True,
-            bufsize=1, # Line buffering
+            bufsize=1,
             universal_newlines=True,
-            preexec_fn=preexec_fn_to_use # Create a new process group
+            preexec_fn=preexec_fn_to_use
         )
 
         with migration_lock:
             migration_state["process"] = process
 
-        # Process output
         if process.stderr:
             for line in iter(process.stderr.readline, ''):
                 line = line.strip()
                 if not line:
                     continue
-                # Update log with verbose output from pg_dump
                 log_message(line, "info")
+
                 if "dumping contents of table" in line:
+                    qualified = None
                     try:
-                        table_name = line.split('"')[1] # Extract table name from quotes
+                        quoted = re.findall(r'"([^"]+)"', line)
+                        if len(quoted) >= 2:
+                            qualified = f"{quoted[-2]}.{quoted[-1]}"
+                        else:
+                            m = re.search(r'dumping contents of table\s+([^\s]+)', line, re.IGNORECASE)
+                            if m:
+                                qualified = m.group(1).strip().strip('"')
+                    except Exception as parse_err:
+                        logger.warning(f"Could not parse table name from dump line: {line} ({parse_err})")
+
+                    if qualified:
                         with migration_lock:
-                            migration_state["progress"]["current_table"] = table_name
+                            migration_state["progress"]["current_table"] = qualified
+                            # Chunk-based progress
+                            chunk_info = migration_state.get("chunk_info_internal")
+                            if chunk_info:
+                                entry = chunk_info["name_to_entry"].get(qualified)
+                                if entry:
+                                    counted = set(migration_state["progress"].get("counted_chunk_names", []))
+                                    names_to_mark = {entry["original_chunk"]}
+                                    if entry["compressed_chunk"]:
+                                        names_to_mark.add(entry["compressed_chunk"])
+                                    # only count once
+                                    if not (counted & names_to_mark):
+                                        migration_state["progress"]["bytes_completed"] += entry["effective_size_bytes"]
+                                        migration_state["progress"]["chunks_completed"] += 1
+                                        counted |= names_to_mark
+                                        migration_state["progress"]["counted_chunk_names"] = list(counted)
+                                    total = migration_state["progress"].get("total_expected_bytes", 0) or 0
+                                    if total > 0:
+                                        migration_state["progress"]["percent_complete"] = min(
+                                            99,
+                                            int((migration_state["progress"]["bytes_completed"] / total) * 100)
+                                        )
+                                    elif migration_state["progress"]["chunks_total"]:
+                                        migration_state["progress"]["percent_complete"] = min(
+                                            99,
+                                            int(migration_state["progress"]["chunks_completed"] * 100 / migration_state["progress"]["chunks_total"])
+                                        )
+                                    elapsed = time.time() - (migration_state.get("start_time") or time.time())
+                                    if elapsed > 5 and migration_state["progress"]["bytes_completed"] > 0:
+                                        rate = migration_state["progress"]["bytes_completed"] / elapsed
+                                        remaining = total - migration_state["progress"]["bytes_completed"]
+                                        eta = int(remaining / rate) if rate > 0 and total > 0 else None
+                                        migration_state["progress"]["estimated_time_remaining"] = eta
+                            # Maintain table count as well (legacy)
                             migration_state["progress"]["tables_completed"] += 1
-                    except IndexError:
-                        logger.warning(f"Could not parse table name from line: {line}")
-                # Check if process is still running
+
                 with migration_lock:
                     if not migration_state["running"]:
-                        break # Stop processing if process was terminated
+                        break
 
-        # Wait for process to complete
         stdout, stderr = process.communicate()
         exit_code = process.returncode
 
         with migration_lock:
-            # Ensure running state is updated based on process completion
-            if migration_state["running"]: # Only update if not stopped manually
+            if migration_state["running"]:
                 if exit_code == 0:
-                    # Get final file size
                     final_size = get_file_size_mb(file_path)
                     migration_state["dump_file_size"] = final_size
                     migration_state["progress"]["current_size_mb"] = round(final_size, 2)
                     migration_state["dump_completed"] = True
                     migration_state["end_time"] = time.time()
+                    migration_state["progress"]["percent_complete"] = 100
+                    migration_state["progress"]["estimated_time_remaining"] = 0
                     total_time = migration_state["end_time"] - migration_state["start_time"]
                     log_message(
                         f"Database dump completed successfully. Size: {round(final_size, 2)} MB. Time: {round(total_time, 2)} seconds",
@@ -253,19 +351,18 @@ def run_dump(source_conn: str, file_path: str, options: dict):
             migration_state["process"] = None
         return False
 
+
 def run_restore(target_conn: str, file_path: str, options: dict):
-    """Run pg_restore in a background thread"""
+    """Run pg_restore in a background thread with chunk-based progress tracking."""
     try:
         if not os.path.exists(file_path):
             log_message(f"Dump file not found: {file_path}", "error")
             with migration_lock:
-                migration_state["running"] = False # Ensure state is consistent
+                migration_state["running"] = False
             return False
 
-        # Set environment variables for connection
         env = os.environ.copy()
 
-        # Run timescaledb_pre_restore() if specified
         if options.get("timescaledb_pre_restore", True):
             pre_restore_cmd = ["psql", target_conn, "-c", "SELECT timescaledb_pre_restore();"]
             log_message("Running timescaledb_pre_restore()", "info", " ".join(pre_restore_cmd))
@@ -280,22 +377,17 @@ def run_restore(target_conn: str, file_path: str, options: dict):
             if pre_restore_process.returncode != 0:
                 log_message(f"Pre-restore failed: {pre_restore_stderr or pre_restore_stdout}", "error")
                 with migration_lock:
-                    migration_state["running"] = False # Ensure state is consistent
+                    migration_state["running"] = False
                 return False
 
-        # Build pg_restore command
         cmd = ["pg_restore", "-d", target_conn, "-v"]
-
-        # Add no-owner flag if specified
         if options.get("no_owner", True):
             cmd.append("--no-owner")
-        # Add clean flag if specified
         if options.get("clean", False):
             cmd.append("--clean")
-        # Add single transaction flag if specified
-        if options.get("single_transaction", True):
+        # Default OFF now:
+        if options.get("single_transaction", False):
             cmd.append("--single-transaction")
-        # Add file path
         cmd.append(file_path)
 
         log_message(f"Starting database restore from {file_path}", "info", " ".join(cmd))
@@ -305,12 +397,9 @@ def run_restore(target_conn: str, file_path: str, options: dict):
             migration_state["running"] = True
             migration_state["operation"] = "restore"
             migration_state["restore_completed"] = False
-            migration_state["progress"]["tables_completed"] = 0 # Reset counter
+            migration_state["progress"]["tables_completed"] = 0
 
-        # Use preexec_fn=os.setsid to create a new process group
-        preexec_fn_to_use = None
-        if hasattr(os, 'setsid'):
-            preexec_fn_to_use = os.setsid
+        preexec_fn_to_use = os.setsid if hasattr(os, 'setsid') else None
 
         process = subprocess.Popen(
             cmd,
@@ -318,51 +407,78 @@ def run_restore(target_conn: str, file_path: str, options: dict):
             stderr=subprocess.PIPE,
             env=env,
             text=True,
-            bufsize=1, # Line buffering
+            bufsize=1,
             universal_newlines=True,
-            preexec_fn=preexec_fn_to_use # Create a new process group
+            preexec_fn=preexec_fn_to_use
         )
 
         with migration_lock:
             migration_state["process"] = process
 
-        # Process output
         if process.stderr:
             for line in iter(process.stderr.readline, ''):
                 line = line.strip()
                 if not line:
                     continue
-                # Log verbose output
                 log_message(line, "info")
-                # Try to parse table name (pg_restore output format varies)
-                if "processing" in line.lower() and ("table data" in line.lower() or "table" in line.lower()):
+
+                if re.search(r'(processing|restoring)\s+data\s+for\s+table', line, re.IGNORECASE):
+                    qualified = None
                     try:
-                        # Attempt to extract table name, might need refinement
-                        parts = line.split()
-                        table_index = -1
-                        if "table" in parts: table_index = parts.index("table") + 1
-                        elif "data" in parts: table_index = parts.index("data") + 1
-
-                        if table_index > 0 and table_index < len(parts):
-                            table_name = parts[table_index].strip('."')
-                            with migration_lock:
-                                migration_state["progress"]["current_table"] = table_name
-                                migration_state["progress"]["tables_completed"] += 1
+                        quoted = re.findall(r'"([^"]+)"', line)
+                        if len(quoted) >= 2:
+                            qualified = f"{quoted[-2]}.{quoted[-1]}"
                         else:
-                             logger.warning(f"Could not parse table name from restore line: {line}")
+                            m = re.search(r'(processing|restoring)\s+data\s+for\s+table\s+([^\s]+)', line, re.IGNORECASE)
+                            if m:
+                                qualified = m.group(2).strip().strip('"')
                     except Exception as parse_err:
                         logger.warning(f"Error parsing restore line '{line}': {parse_err}")
-                # Check if process is still running
+
+                    if qualified:
+                        with migration_lock:
+                            migration_state["progress"]["current_table"] = qualified
+                            chunk_info = migration_state.get("chunk_info_internal")
+                            if chunk_info:
+                                entry = chunk_info["name_to_entry"].get(qualified)
+                                if entry:
+                                    counted = set(migration_state["progress"].get("counted_chunk_names", []))
+                                    names_to_mark = {entry["original_chunk"]}
+                                    if entry["compressed_chunk"]:
+                                        names_to_mark.add(entry["compressed_chunk"])
+                                    if not (counted & names_to_mark):
+                                        migration_state["progress"]["bytes_completed"] += entry["effective_size_bytes"]
+                                        migration_state["progress"]["chunks_completed"] += 1
+                                        counted |= names_to_mark
+                                        migration_state["progress"]["counted_chunk_names"] = list(counted)
+                                    total = migration_state["progress"].get("total_expected_bytes", 0) or 0
+                                    if total > 0:
+                                        migration_state["progress"]["percent_complete"] = min(
+                                            99,
+                                            int((migration_state["progress"]["bytes_completed"] / total) * 100)
+                                        )
+                                    elif migration_state["progress"]["chunks_total"]:
+                                        migration_state["progress"]["percent_complete"] = min(
+                                            99,
+                                            int(migration_state["progress"]["chunks_completed"] * 100 / migration_state["progress"]["chunks_total"])
+                                        )
+                                    elapsed = time.time() - (migration_state.get("start_time") or time.time())
+                                    if elapsed > 5 and migration_state["progress"]["bytes_completed"] > 0:
+                                        rate = migration_state["progress"]["bytes_completed"] / elapsed
+                                        remaining = total - migration_state["progress"]["bytes_completed"]
+                                        eta = int(remaining / rate) if rate > 0 and total > 0 else None
+                                        migration_state["progress"]["estimated_time_remaining"] = eta
+                            # Maintain legacy counter
+                            migration_state["progress"]["tables_completed"] += 1
+
                 with migration_lock:
                     if not migration_state["running"]:
-                        break # Stop processing if process was terminated
+                        break
 
-        # Wait for process to complete
         stdout, stderr = process.communicate()
         exit_code = process.returncode
 
         post_restore_success = True
-        # Run timescaledb_post_restore() if specified and restore was successful so far
         if exit_code == 0 and options.get("timescaledb_post_restore", True):
             post_restore_cmd = ["psql", target_conn, "-c", "SELECT timescaledb_post_restore(); ANALYZE;"]
             log_message("Running timescaledb_post_restore() and ANALYZE", "info", " ".join(post_restore_cmd))
@@ -376,14 +492,15 @@ def run_restore(target_conn: str, file_path: str, options: dict):
             post_restore_stdout, post_restore_stderr = post_restore_process.communicate()
             if post_restore_process.returncode != 0:
                 log_message(f"Post-restore failed: {post_restore_stderr or post_restore_stdout}", "error")
-                post_restore_success = False # Mark post-restore as failed
+                post_restore_success = False
 
         with migration_lock:
-            # Ensure running state is updated based on process completion
-            if migration_state["running"]: # Only update if not stopped manually
+            if migration_state["running"]:
                 if exit_code == 0 and post_restore_success:
                     migration_state["restore_completed"] = True
                     migration_state["end_time"] = time.time()
+                    migration_state["progress"]["percent_complete"] = 100
+                    migration_state["progress"]["estimated_time_remaining"] = 0
                     total_time = migration_state["end_time"] - migration_state["start_time"]
                     log_message(
                         f"Database restore completed successfully. Time: {round(total_time, 2)} seconds",
@@ -392,7 +509,6 @@ def run_restore(target_conn: str, file_path: str, options: dict):
                 elif exit_code != 0:
                     error_message = stderr or stdout or "Unknown error during restore"
                     log_message(f"Database restore failed: {error_message}", "error")
-                # If post_restore failed, it's already logged.
 
                 migration_state["running"] = False
                 migration_state["process"] = None
@@ -406,6 +522,7 @@ def run_restore(target_conn: str, file_path: str, options: dict):
             migration_state["process"] = None
         return False
 
+
 # Replace the old stop_current_process with the new one
 def stop_current_process():
     """Stop the current process with improved forceful termination"""
@@ -2305,7 +2422,7 @@ async def home(request: Request):
                                     <label class="form-label" for="clean">Clean (drop) database objects before recreating (--clean)</label>
                                 </div>
                                 <div class="form-check mb-2">
-                                    <input class="form-check-input" type="checkbox" id="single-transaction" checked>
+                                    <input class="form-check-input" type="checkbox" id="single-transaction">
                                     <label class="form-label" for="single-transaction">Restore as a single transaction (--single-transaction)</label>
                                 </div>
                             </div>
@@ -2396,12 +2513,18 @@ async def home(request: Request):
                     <div class="logs-container mt-4">
                         <div class="logs-header">
                             <div class="logs-title">Activity Log</div>
-                            <div class="logs-filters">
-                                <button class="log-filter active" data-level="all">All</button>
-                                <button class="log-filter" data-level="info">Info</button>
-                                <button class="log-filter" data-level="success">Success</button>
-                                <button class="log-filter" data-level="warning">Warning</button>
-                                <button class="log-filter" data-level="error">Error</button>
+                            <div class="d-flex align-center gap-3">
+                                <div class="logs-filters">
+                                    <button class="log-filter active" data-level="all">All</button>
+                                    <button class="log-filter" data-level="info">Info</button>
+                                    <button class="log-filter" data-level="success">Success</button>
+                                    <button class="log-filter" data-level="warning">Warning</button>
+                                    <button class="log-filter" data-level="error">Error</button>
+                                </div>
+                                <div class="form-check" title="When enabled, the log viewer will not auto-scroll to new entries">
+                                    <input class="form-check-input" type="checkbox" id="scroll-lock">
+                                    <label class="form-label" for="scroll-lock">Scroll Lock</label>
+                                </div>
                             </div>
                         </div>
                         <div class="logs-body" id="logs-output">
@@ -2547,6 +2670,24 @@ async def home(request: Request):
             const cancelConfirmBtn = document.getElementById('cancel-confirm-btn');
             const confirmActionBtn = document.getElementById('confirm-action-btn');
             const confirmModalBody = document.getElementById('confirm-modal-body');
+
+            const scrollLockCheckbox = document.getElementById('scroll-lock');
+            let scrollLock = JSON.parse(localStorage.getItem('scroll_lock') || 'false');
+            if (scrollLockCheckbox) {
+                scrollLockCheckbox.checked = scrollLock;
+            }
+            function maybeAutoScroll(el) {
+                if (!scrollLock && el) {
+                    el.scrollTop = el.scrollHeight;
+                }
+            }
+            if (scrollLockCheckbox) {
+                scrollLockCheckbox.addEventListener('change', () => {
+                    scrollLock = scrollLockCheckbox.checked;
+                    localStorage.setItem('scroll_lock', JSON.stringify(scrollLock));
+                });
+            }
+
             // State variables
             let sizeChart = null;
             let updateInterval = null;
@@ -2642,8 +2783,11 @@ async def home(request: Request):
                     tab.classList.add('active');
                     const tabId = tab.getAttribute('data-tab');
                     document.getElementById(`${tabId}-tab`).classList.add('active');
+                    // Persist active tab
+                    localStorage.setItem('active_tab', tabId);
                 });
             });
+
             // Log filter functionality
             logFilters.forEach(filter => {
                 filter.addEventListener('click', () => {
@@ -3390,7 +3534,7 @@ async def home(request: Request):
 
                 terminalOutput.appendChild(line);
                 // Auto-scroll
-                terminalOutput.scrollTop = terminalOutput.scrollHeight;
+                maybeAutoScroll(terminalOutput);
             }
             // Add log entry
             function addLogEntry(log) {
@@ -3416,7 +3560,7 @@ async def home(request: Request):
                     logEntry.style.display = 'none';
                 }
                 // Scroll to bottom
-                logsOutput.scrollTop = logsOutput.scrollHeight;
+                maybeAutoScroll(logsOutput);
 
                 // Also add to terminal, map log level to terminal type
                 let terminalType = 'output';
@@ -3695,6 +3839,41 @@ async def home(request: Request):
                 await updateStatus(); // This also updates migration_state
                 // Add initial log check
                 await checkForNewLogs();
+                const savedTab = localStorage.getItem('active_tab');
+                if (savedTab && document.querySelector(`.tab[data-tab="${savedTab}"]`)) {
+                    document.querySelector(`.tab[data-tab="${savedTab}"]`).click();
+                }
+                // Reflect completed dump state on reload
+                if (migration_state && migration_state.dump_file) {
+                    dumpProgressSection.classList.remove('hidden');
+                    dumpFilePathElement.textContent = migration_state.dump_file;
+                    currentSizeElement.textContent = `${(migration_state.dump_file_size || 0).toFixed(2)} MB`;
+                    if (migration_state.dump_completed) {
+                        dumpStatusElement.textContent = 'Completed';
+                        dumpStatusElement.className = 'badge success';
+                        const downloadPath = `/downloads/${migration_state.dump_file.split(/[\\/]/).pop()}`;
+                        downloadDumpBtn.href = downloadPath;
+                        downloadDumpBtn.disabled = false;
+                        gotoRestoreFromDumpBtn.disabled = false;
+                        statusBadge.className = 'status-badge success';
+                        statusBadge.textContent = 'Dump Complete';
+                    } else if (migration_state.running && migration_state.operation === 'dump') {
+                        dumpStatusElement.textContent = 'In Progress';
+                        dumpStatusElement.className = 'badge info';
+                    }
+                }
+                
+                // Reflect completed restore state on reload
+                if (migration_state && migration_state.restore_completed) {
+                    restoreProgressSection.classList.remove('hidden');
+                    restoreStatusElement.textContent = 'Completed';
+                    restoreSubstatusElement.textContent = 'Restore completed successfully';
+                    restoreProgressBar.style.width = '100%';
+                    restoreProgressBar.classList.remove('animated');
+                    restoreProgressValue.textContent = '100%';
+                    statusBadge.className = 'status-badge success';
+                    statusBadge.textContent = 'Restore Complete';
+                }
 
                  // If a process was running when the page loaded, sync UI and start updates
                  if (migration_state && migration_state.running) {
@@ -3715,6 +3894,7 @@ async def home(request: Request):
                          startDumpBtn.disabled = true;
                      }
                  }
+                 
             }
             // Initialize app
             initialize();
@@ -3769,82 +3949,91 @@ async def home(request: Request):
 
 @app.post("/test-connection")
 async def test_connection_endpoint(data: Dict[str, str]):
-    """Test a database connection and get basic info"""
+    """Test a database connection and get basic info; if source, pre-load TimescaleDB chunk map."""
     try:
         connection_string = data.get("connection_string")
-        connection_type = data.get("connection_type", "source") # Added type
+        connection_type = data.get("connection_type", "source")
         if not connection_string:
             return JSONResponse(
                 status_code=400,
                 content={"success": False, "message": "Connection string is required"}
             )
 
-        # Test the connection using internal logic
         if not test_connection_logic(connection_string):
-            # Error is logged within test_connection_logic
             return JSONResponse(
                 content={"success": False, "message": "Failed to connect to database"}
             )
 
-        # If connection successful, get database info
         conn = psycopg2.connect(connection_string)
         try:
+            chunk_summary_resp = None
             with conn.cursor() as cur:
-                # Get server info
                 cur.execute("SELECT version()")
                 version_result = cur.fetchone()
                 version = version_result[0] if version_result else "Unknown"
 
-                # Check if TimescaleDB is installed and get version
+                # Check TimescaleDB
                 is_timescaledb = False
                 ts_version = None
                 try:
-                    # Use EXISTS for better performance and error handling if extension not present
                     cur.execute("SELECT EXISTS (SELECT 1 FROM pg_extension WHERE extname = 'timescaledb');")
                     if cur.fetchone()[0]:
-                         cur.execute("SELECT extversion FROM pg_extension WHERE extname = 'timescaledb'")
-                         ts_version_result = cur.fetchone()
-                         is_timescaledb = ts_version_result is not None
-                         ts_version = ts_version_result[0] if ts_version_result else None
+                        cur.execute("SELECT extversion FROM pg_extension WHERE extname = 'timescaledb'")
+                        ts_version_result = cur.fetchone()
+                        is_timescaledb = ts_version_result is not None
+                        ts_version = ts_version_result[0] if ts_version_result else None
                 except psycopg2.Error as ts_err:
-                     logger.warning(f"Could not check TimescaleDB extension: {ts_err}")
-                     # Continue without TimescaleDB info if query fails
+                    logger.warning(f"Could not check TimescaleDB extension: {ts_err}")
 
-                # Get database name
                 cur.execute("SELECT current_database()")
                 db_result = cur.fetchone()
                 database = db_result[0] if db_result else "Unknown"
 
-                # Extract server host/IP (best effort parsing)
                 server_match = "Unknown"
                 try:
-                    # Extract from connection string if possible (more reliable)
                     host_part = connection_string.split('@')[-1].split('/')[0].split(':')[0]
                     if host_part:
                         server_match = host_part
-                    # Fallback to parsing version string if needed
                     elif " on " in version:
-                         server_match = version.split(" on ")[-1].split(",")[0]
+                        server_match = version.split(" on ")[-1].split(",")[0]
                 except Exception:
                     logger.warning("Could not parse server host from connection string or version.")
 
+                # If it's the source and TimescaleDB is present, build and cache chunk info
+                if connection_type == "source" and is_timescaledb:
+                    chunk_info = build_timescaledb_chunk_info(conn)
+                    with migration_lock:
+                        migration_state["chunk_info_internal"] = chunk_info
+                    if chunk_info and chunk_info.get("summary"):
+                        total_mb = round(chunk_info["summary"]["total_effective_size_bytes"] / (1024 * 1024), 2)
+                        chunk_summary_resp = {
+                            "chunks_count": chunk_info["summary"]["chunks_count"],
+                            "hypertables_count": chunk_info["summary"]["hypertables_count"],
+                            "total_effective_size_mb": total_mb
+                        }
+                        log_message(
+                            f"Loaded TimescaleDB chunk map: {chunk_summary_resp['chunks_count']} chunks across "
+                            f"{chunk_summary_resp['hypertables_count']} hypertables, ~{total_mb} MB effective data.",
+                            "info"
+                        )
 
                 log_message(f"Successful connection test to {connection_type} database: {database} on {server_match}", "success")
                 return JSONResponse(content={
                     "success": True,
                     "version": version,
                     "is_timescaledb": is_timescaledb,
-                    "timescaledb_version": ts_version, # Add this line
+                    "timescaledb_version": ts_version,
                     "database": database,
-                    "server": server_match
+                    "server": server_match,
+                    "chunk_summary": chunk_summary_resp  # May be None if not source or no TSDB
                 })
         finally:
             conn.close()
     except psycopg2.Error as db_err:
-         log_message(f"Database connection error during info fetch: {str(db_err)}", "error")
-         return JSONResponse(
-             content={"success": False, "message": f"Database error: {str(db_err)}"}
-         )
+        log_message(f"Database connection error during info fetch: {str(db_err)}", "error")
+        return JSONResponse(
+            content={"success": False, "message": f"Database error: {str(db_err)}"}
+        )
     except Exception as e:
         log_message(f"Connection test failed unexpectedly: {str(e)}", "error")
         return JSONResponse(
@@ -3852,6 +4041,7 @@ async def test_connection_endpoint(data: Dict[str, str]):
             content={"success": False, "message": f"An unexpected error occurred: {str(e)}"}
         )
 
+
 @app.post("/database-info")
 async def get_database_info(data: Dict[str, str]):
     """Get additional database information like table count and size"""
@@ -3921,60 +4111,57 @@ async def start_dump(data: Dict[str, Any], background_tasks: BackgroundTasks):
                 content={"success": False, "message": "Source connection string is required"}
             )
 
-        # Basic validation: Test connection before starting dump
         if not test_connection_logic(source_conn):
-             return JSONResponse(
+            return JSONResponse(
                 status_code=400,
                 content={"success": False, "message": "Source connection failed. Cannot start dump."}
             )
 
-        # Stop any running process first (important!)
         if migration_state["running"]:
             logger.warning("Another process is running. Stopping it before starting dump.")
             stopped = stop_current_process()
             if not stopped:
-                 logger.error("Failed to stop the existing process. Cannot start dump.")
-                 return JSONResponse(
-                    status_code=500, # Internal Server Error might be appropriate
+                logger.error("Failed to stop the existing process. Cannot start dump.")
+                return JSONResponse(
+                    status_code=500,
                     content={"success": False, "message": "Failed to stop the currently running process."}
                 )
-            # Add a small delay to allow the process to fully terminate
             time.sleep(0.5)
 
-
-        # Create dump file path
         filename = options.get("filename", "timescale_backup").strip()
-        # Basic filename sanitization (replace spaces, avoid path traversal)
         filename = filename.replace(" ", "_").replace("..", "").replace("/", "").replace("\\", "")
-        if not filename: filename = "timescale_backup" # Fallback if sanitization results in empty name
+        if not filename:
+            filename = "timescale_backup"
 
         format_flag = options.get("format", "c")
-
-        # Determine file extension
         extension = ".dump"
         if format_flag == "p":
             extension = ".sql"
         elif format_flag == "d":
-            extension = ""  # Directory format has no extension
+            extension = ""
         elif format_flag == "t":
             extension = ".tar"
 
-        # Generate file path carefully
-        dumps_dir = Path("dumps").resolve() # Ensure absolute path
+        dumps_dir = Path("dumps").resolve()
         file_path = dumps_dir / f"{filename}{extension}"
-
-        # Prevent potential directory traversal if filename somehow still contains harmful chars
         if not str(file_path).startswith(str(dumps_dir)):
-             logger.error(f"Invalid filename resulted in path traversal attempt: {filename}")
-             return JSONResponse(
+            logger.error(f"Invalid filename resulted in path traversal attempt: {filename}")
+            return JSONResponse(
                 status_code=400,
                 content={"success": False, "message": "Invalid filename specified."}
             )
 
-        # Reset state before starting background task
+        # Read expected totals from chunk info (if any loaded from source test)
         with migration_lock:
-            migration_state["id"] = str(uuid.uuid4()) # New ID for new operation
-            migration_state["running"] = False # Will be set true by run_dump
+            chunk_info = migration_state.get("chunk_info_internal")
+            expected_bytes = 0
+            chunks_total = 0
+            if chunk_info and "summary" in chunk_info:
+                expected_bytes = int(chunk_info["summary"].get("total_effective_size_bytes", 0) or 0)
+                chunks_total = int(chunk_info["summary"].get("chunks_count", 0) or 0)
+
+            migration_state["id"] = str(uuid.uuid4())
+            migration_state["running"] = False
             migration_state["operation"] = "dump"
             migration_state["start_time"] = None
             migration_state["end_time"] = None
@@ -3984,29 +4171,31 @@ async def start_dump(data: Dict[str, Any], background_tasks: BackgroundTasks):
             migration_state["dump_completed"] = False
             migration_state["restore_completed"] = False
             migration_state["last_activity"] = time.time()
-            # Keep logs or clear them? Let's keep them for now.
-            # migration_state["log"] = []
             migration_state["process"] = None
-            migration_state["progress"] = { # Reset progress
+            migration_state["progress"] = {
                 "current_table": None,
                 "tables_completed": 0,
                 "total_tables": 0,
                 "current_size_mb": 0,
                 "growth_rate_mb_per_sec": 0,
                 "estimated_time_remaining": None,
-                "percent_complete": 0
+                "percent_complete": 0,
+                "total_expected_bytes": expected_bytes,
+                "bytes_completed": 0,
+                "chunks_completed": 0,
+                "chunks_total": chunks_total,
+                "counted_chunk_names": []
             }
 
+        if expected_bytes > 0:
+            log_message(f"Using chunk map for dump progress: {round(expected_bytes / (1024*1024), 2)} MB across {chunks_total} chunks.", "info")
 
-        # Start dump in background
         background_tasks.add_task(run_dump, source_conn, str(file_path), options)
 
-        # Create command preview (with redacted password)
-        # Redact password more carefully
         try:
             source_safe_preview = source_conn.replace(source_conn.split('://')[1].split(':')[1].split('@')[0], '***')
         except:
-            source_safe_preview = "postgres://user:***@host/db" # Fallback preview
+            source_safe_preview = "postgres://user:***@host/db"
 
         cmd_preview = f'"{source_safe_preview}" -F{format_flag} -v'
         if options.get("compression") and options["compression"] != "default":
@@ -4015,7 +4204,6 @@ async def start_dump(data: Dict[str, Any], background_tasks: BackgroundTasks):
             cmd_preview += f' -n "{options["schema"]}"'
         cmd_preview += f' -f "{os.path.basename(file_path)}"'
 
-
         return JSONResponse(content={
             "success": True,
             "message": "Dump process initiated",
@@ -4029,6 +4217,7 @@ async def start_dump(data: Dict[str, Any], background_tasks: BackgroundTasks):
             content={"success": False, "message": f"An unexpected error occurred: {str(e)}"}
         )
 
+
 @app.post("/start-restore")
 async def start_restore(data: Dict[str, Any], background_tasks: BackgroundTasks):
     """Start a database restore process"""
@@ -4048,75 +4237,84 @@ async def start_restore(data: Dict[str, Any], background_tasks: BackgroundTasks)
                 content={"success": False, "message": "Dump file is required"}
             )
 
-        # Basic validation: Test connection before starting restore
         if not test_connection_logic(target_conn):
-             return JSONResponse(
+            return JSONResponse(
                 status_code=400,
                 content={"success": False, "message": "Target connection failed. Cannot start restore."}
             )
 
-        # Validate dump file path exists and is within the dumps directory
         dumps_dir = Path("dumps").resolve()
         dump_file_path = Path(dump_file).resolve()
         if not dump_file_path.exists() or not str(dump_file_path).startswith(str(dumps_dir)):
-             logger.error(f"Invalid or non-existent dump file specified: {dump_file}")
-             return JSONResponse(
+            logger.error(f"Invalid or non-existent dump file specified: {dump_file}")
+            return JSONResponse(
                 status_code=400,
                 content={"success": False, "message": "Invalid or non-existent dump file selected."}
             )
 
-        # Stop any running process first (important!)
         if migration_state["running"]:
             logger.warning("Another process is running. Stopping it before starting restore.")
             stopped = stop_current_process()
             if not stopped:
-                 logger.error("Failed to stop the existing process. Cannot start restore.")
-                 return JSONResponse(
+                logger.error("Failed to stop the existing process. Cannot start restore.")
+                return JSONResponse(
                     status_code=500,
                     content={"success": False, "message": "Failed to stop the currently running process."}
                 )
-            time.sleep(0.5) # Allow time for termination
+            time.sleep(0.5)
 
-        # Reset state before starting background task
         with migration_lock:
-            migration_state["id"] = str(uuid.uuid4()) # New ID for new operation
-            migration_state["running"] = False # Will be set true by run_restore
+            chunk_info = migration_state.get("chunk_info_internal")
+            expected_bytes = 0
+            chunks_total = 0
+            if chunk_info and "summary" in chunk_info:
+                expected_bytes = int(chunk_info["summary"].get("total_effective_size_bytes", 0) or 0)
+                chunks_total = int(chunk_info["summary"].get("chunks_count", 0) or 0)
+
+            migration_state["id"] = str(uuid.uuid4())
+            migration_state["running"] = False
             migration_state["operation"] = "restore"
             migration_state["start_time"] = None
             migration_state["end_time"] = None
-            migration_state["dump_file"] = None # Not relevant for restore state itself
+            migration_state["dump_file"] = None
             migration_state["dump_file_size"] = 0
             migration_state["previous_size"] = 0
             migration_state["dump_completed"] = False
             migration_state["restore_completed"] = False
             migration_state["last_activity"] = time.time()
-            # Keep logs
             migration_state["process"] = None
-            migration_state["progress"] = { # Reset progress
+            migration_state["progress"] = {
                 "current_table": None,
                 "tables_completed": 0,
-                "total_tables": 0, # We don't easily know this for restore
+                "total_tables": 0,
                 "current_size_mb": 0,
                 "growth_rate_mb_per_sec": 0,
                 "estimated_time_remaining": None,
-                "percent_complete": 0
+                "percent_complete": 0,
+                "total_expected_bytes": expected_bytes,
+                "bytes_completed": 0,
+                "chunks_completed": 0,
+                "chunks_total": chunks_total,
+                "counted_chunk_names": []
             }
 
-        # Start restore in background
+        if expected_bytes > 0:
+            log_message(f"Using chunk map for restore progress: {round(expected_bytes / (1024*1024), 2)} MB across {chunks_total} chunks.", "info")
+
         background_tasks.add_task(run_restore, target_conn, str(dump_file_path), options)
 
-        # Create command preview (with redacted password)
         try:
             target_safe_preview = target_conn.replace(target_conn.split('://')[1].split(':')[1].split('@')[0], '***')
         except:
-             target_safe_preview = "postgres://user:***@host/db"
+            target_safe_preview = "postgres://user:***@host/db"
 
         cmd_preview = f'-d "{target_safe_preview}" -v'
         if options.get("no_owner", True):
             cmd_preview += " --no-owner"
         if options.get("clean", False):
             cmd_preview += " --clean"
-        if options.get("single_transaction", True):
+        # Default OFF now for single-transaction
+        if options.get("single_transaction", False):
             cmd_preview += " --single-transaction"
         cmd_preview += f' "{os.path.basename(dump_file)}"'
 
@@ -4132,6 +4330,7 @@ async def start_restore(data: Dict[str, Any], background_tasks: BackgroundTasks)
             content={"success": False, "message": f"An unexpected error occurred: {str(e)}"}
         )
 
+
 @app.post("/stop-process")
 async def stop_process_endpoint():
     """Stop the current database process"""
@@ -4171,18 +4370,33 @@ async def stop_process_endpoint():
 
 @app.get("/status")
 async def get_status():
-    """Get the current migration status"""
-    # Return a copy to avoid potential modification issues if state grows complex
+    """Get the current migration status (without heavy internal structures)."""
     with migration_lock:
         state_copy = migration_state.copy()
-        # Ensure process object is not sent over JSON
         state_copy["process"] = None
-        # Optionally limit log size sent back if it gets large
-        # MAX_LOGS_IN_STATUS = 100
-        # if len(state_copy["log"]) > MAX_LOGS_IN_STATUS:
-        #     state_copy["log"] = state_copy["log"][-MAX_LOGS_IN_STATUS:]
+
+        # Remove heavy internal chunk details; expose a small summary instead
+        if state_copy.get("chunk_info_internal"):
+            chunk_info = state_copy["chunk_info_internal"]
+            summary = chunk_info.get("summary", {}) if chunk_info else {}
+            if "total_effective_size_bytes" in summary:
+                total_mb = round(summary["total_effective_size_bytes"] / (1024 * 1024), 2)
+            else:
+                total_mb = None
+            state_copy["chunk_summary"] = {
+                "chunks_count": summary.get("chunks_count"),
+                "hypertables_count": summary.get("hypertables_count"),
+                "total_effective_size_mb": total_mb
+            }
+            del state_copy["chunk_info_internal"]
+
+        # Trim internal counted names (can be large)
+        if "progress" in state_copy and "counted_chunk_names" in state_copy["progress"]:
+            state_copy["progress"]["counted_chunk_names"] = []
+
     return state_copy
 
+
 @app.post("/clear-logs")
 async def clear_logs():
     """Clear all logs"""