Compare commits
7 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| eeacbfa007 | |||
| 7711a206ab | |||
| ba6e8a2b39 | |||
| ec5e89eab7 | |||
| e24d7ab49f | |||
| 721e53fe6a | |||
| 4e09066aa5 |
72
CHANGELOG.md
72
CHANGELOG.md
@@ -5,6 +5,78 @@ All notable changes to dbbackup will be documented in this file.
|
||||
The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
|
||||
and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
|
||||
|
||||
## [3.42.35] - 2026-01-15 "TUI Detailed Progress"
|
||||
|
||||
### Added - Enhanced TUI Progress Display
|
||||
- **Detailed progress bar in TUI restore** - schollz-style progress bar with:
|
||||
- Byte progress display (e.g., `245 MB / 1.2 GB`)
|
||||
- Transfer speed calculation (e.g., `45 MB/s`)
|
||||
- ETA prediction for long operations
|
||||
- Unicode block-based visual bar
|
||||
- **Real-time extraction progress** - Archive extraction now reports actual bytes processed
|
||||
- **Go-native tar extraction** - Uses Go's `archive/tar` + `compress/gzip` when progress callback is set
|
||||
- **New `DetailedProgress` component** in TUI package:
|
||||
- `NewDetailedProgress(total, description)` - Byte-based progress
|
||||
- `NewDetailedProgressItems(total, description)` - Item count progress
|
||||
- `NewDetailedProgressSpinner(description)` - Indeterminate spinner
|
||||
- `RenderProgressBar(width)` - Generate schollz-style output
|
||||
- **Progress callback API** in restore engine:
|
||||
- `SetProgressCallback(func(current, total int64, description string))`
|
||||
- Allows TUI to receive real-time progress updates from restore operations
|
||||
- **Shared progress state** pattern for Bubble Tea integration
|
||||
|
||||
### Changed
|
||||
- TUI restore execution now shows detailed byte progress during archive extraction
|
||||
- Cluster restore shows extraction progress instead of just spinner
|
||||
- Falls back to shell `tar` command when no progress callback is set (faster)
|
||||
|
||||
### Technical Details
|
||||
- `progressReader` wrapper tracks bytes read through gzip/tar pipeline
|
||||
- Throttled progress updates (every 100ms) to avoid UI flooding
|
||||
- Thread-safe shared state pattern for cross-goroutine progress updates
|
||||
|
||||
## [3.42.34] - 2026-01-14 "Filesystem Abstraction"
|
||||
|
||||
### Added - spf13/afero for Filesystem Abstraction
|
||||
- **New `internal/fs` package** for testable filesystem operations
|
||||
- **In-memory filesystem** for unit testing without disk I/O
|
||||
- **Global FS interface** that can be swapped for testing:
|
||||
```go
|
||||
fs.SetFS(afero.NewMemMapFs()) // Use memory
|
||||
fs.ResetFS() // Back to real disk
|
||||
```
|
||||
- **Wrapper functions** for all common file operations:
|
||||
- `ReadFile`, `WriteFile`, `Create`, `Open`, `Remove`, `RemoveAll`
|
||||
- `Mkdir`, `MkdirAll`, `ReadDir`, `Walk`, `Glob`
|
||||
- `Exists`, `DirExists`, `IsDir`, `IsEmpty`
|
||||
- `TempDir`, `TempFile`, `CopyFile`, `FileSize`
|
||||
- **Testing helpers**:
|
||||
- `WithMemFs(fn)` - Execute function with temp in-memory FS
|
||||
- `SetupTestDir(files)` - Create test directory structure
|
||||
- **Comprehensive test suite** demonstrating usage
|
||||
|
||||
### Changed
|
||||
- Upgraded afero from v1.10.0 to v1.15.0
|
||||
|
||||
## [3.42.33] - 2026-01-14 "Exponential Backoff Retry"
|
||||
|
||||
### Added - cenkalti/backoff for Cloud Operation Retry
|
||||
- **Exponential backoff retry** for all cloud operations (S3, Azure, GCS)
|
||||
- **Retry configurations**:
|
||||
- `DefaultRetryConfig()` - 5 retries, 500ms→30s backoff, 5 min max
|
||||
- `AggressiveRetryConfig()` - 10 retries, 1s→60s backoff, 15 min max
|
||||
- `QuickRetryConfig()` - 3 retries, 100ms→5s backoff, 30s max
|
||||
- **Smart error classification**:
|
||||
- `IsPermanentError()` - Auth/bucket errors (no retry)
|
||||
- `IsRetryableError()` - Timeout/network errors (retry)
|
||||
- **Retry logging** - Each retry attempt is logged with wait duration
|
||||
|
||||
### Changed
|
||||
- S3 simple upload, multipart upload, download now retry on transient failures
|
||||
- Azure simple upload, download now retry on transient failures
|
||||
- GCS upload, download now retry on transient failures
|
||||
- Large file multipart uploads use `AggressiveRetryConfig()` (more retries)
|
||||
|
||||
## [3.42.32] - 2026-01-14 "Cross-Platform Colors"
|
||||
|
||||
### Added - fatih/color for Cross-Platform Terminal Colors
|
||||
|
||||
@@ -56,7 +56,7 @@ Download from [releases](https://git.uuxo.net/UUXO/dbbackup/releases):
|
||||
|
||||
```bash
|
||||
# Linux x86_64
|
||||
wget https://git.uuxo.net/UUXO/dbbackup/releases/download/v3.42.1/dbbackup-linux-amd64
|
||||
wget https://git.uuxo.net/UUXO/dbbackup/releases/download/v3.42.35/dbbackup-linux-amd64
|
||||
chmod +x dbbackup-linux-amd64
|
||||
sudo mv dbbackup-linux-amd64 /usr/local/bin/dbbackup
|
||||
```
|
||||
|
||||
@@ -3,9 +3,9 @@
|
||||
This directory contains pre-compiled binaries for the DB Backup Tool across multiple platforms and architectures.
|
||||
|
||||
## Build Information
|
||||
- **Version**: 3.42.31
|
||||
- **Build Time**: 2026-01-14_15:07:12_UTC
|
||||
- **Git Commit**: dc6dfd8
|
||||
- **Version**: 3.42.34
|
||||
- **Build Time**: 2026-01-14_16:19:00_UTC
|
||||
- **Git Commit**: 7711a20
|
||||
|
||||
## Recent Updates (v1.1.0)
|
||||
- ✅ Fixed TUI progress display with line-by-line output
|
||||
|
||||
2
go.mod
2
go.mod
@@ -57,6 +57,7 @@ require (
|
||||
github.com/aws/aws-sdk-go-v2/service/sts v1.41.2 // indirect
|
||||
github.com/aws/smithy-go v1.23.2 // indirect
|
||||
github.com/aymanbagabas/go-osc52/v2 v2.0.1 // indirect
|
||||
github.com/cenkalti/backoff/v4 v4.3.0 // indirect
|
||||
github.com/cespare/xxhash/v2 v2.3.0 // indirect
|
||||
github.com/charmbracelet/colorprofile v0.2.3-0.20250311203215-f60798e515dc // indirect
|
||||
github.com/charmbracelet/x/ansi v0.10.1 // indirect
|
||||
@@ -96,6 +97,7 @@ require (
|
||||
github.com/power-devops/perfstat v0.0.0-20210106213030-5aafc221ea8c // indirect
|
||||
github.com/rivo/uniseg v0.4.7 // indirect
|
||||
github.com/schollz/progressbar/v3 v3.19.0 // indirect
|
||||
github.com/spf13/afero v1.15.0 // indirect
|
||||
github.com/spiffe/go-spiffe/v2 v2.5.0 // indirect
|
||||
github.com/tklauser/go-sysconf v0.3.12 // indirect
|
||||
github.com/tklauser/numcpus v0.6.1 // indirect
|
||||
|
||||
4
go.sum
4
go.sum
@@ -84,6 +84,8 @@ github.com/aws/smithy-go v1.23.2 h1:Crv0eatJUQhaManss33hS5r40CG3ZFH+21XSkqMrIUM=
|
||||
github.com/aws/smithy-go v1.23.2/go.mod h1:LEj2LM3rBRQJxPZTB4KuzZkaZYnZPnvgIhb4pu07mx0=
|
||||
github.com/aymanbagabas/go-osc52/v2 v2.0.1 h1:HwpRHbFMcZLEVr42D4p7XBqjyuxQH5SMiErDT4WkJ2k=
|
||||
github.com/aymanbagabas/go-osc52/v2 v2.0.1/go.mod h1:uYgXzlJ7ZpABp8OJ+exZzJJhRNQ2ASbcXHWsFqH8hp8=
|
||||
github.com/cenkalti/backoff/v4 v4.3.0 h1:MyRJ/UdXutAwSAT+s3wNd7MfTIcy71VQueUuFK343L8=
|
||||
github.com/cenkalti/backoff/v4 v4.3.0/go.mod h1:Y3VNntkOUPxTVeUxJ/G5vcM//AlwfmyYozVcomhLiZE=
|
||||
github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
|
||||
github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
|
||||
github.com/charmbracelet/bubbles v0.21.0 h1:9TdC97SdRVg/1aaXNVWfFH3nnLAwOXr8Fn6u6mfQdFs=
|
||||
@@ -209,6 +211,8 @@ github.com/shirou/gopsutil/v3 v3.24.5 h1:i0t8kL+kQTvpAYToeuiVk3TgDeKOFioZO3Ztz/i
|
||||
github.com/shirou/gopsutil/v3 v3.24.5/go.mod h1:bsoOS1aStSs9ErQ1WWfxllSeS1K5D+U30r2NfcubMVk=
|
||||
github.com/sirupsen/logrus v1.9.3 h1:dueUQJ1C2q9oE3F7wvmSGAaVtTmUizReu6fjN8uqzbQ=
|
||||
github.com/sirupsen/logrus v1.9.3/go.mod h1:naHLuLoDiP4jHNo9R0sCBMtWGeIprob74mVsIT4qYEQ=
|
||||
github.com/spf13/afero v1.15.0 h1:b/YBCLWAJdFWJTN9cLhiXXcD7mzKn9Dm86dNnfyQw1I=
|
||||
github.com/spf13/afero v1.15.0/go.mod h1:NC2ByUVxtQs4b3sIUphxK0NioZnmxgyCrfzeuq8lxMg=
|
||||
github.com/spf13/cobra v1.10.1 h1:lJeBwCfmrnXthfAupyUTzJ/J4Nc1RsHC/mSRU2dll/s=
|
||||
github.com/spf13/cobra v1.10.1/go.mod h1:7SmJGaTHFVBY0jW4NXGluQoLvhqFQM+6XSKD+P4XaB0=
|
||||
github.com/spf13/pflag v1.0.9 h1:9exaQaMOCwffKiiiYk6/BndUBv+iRViNW+4lEMi0PvY=
|
||||
|
||||
@@ -151,8 +151,14 @@ func (a *AzureBackend) Upload(ctx context.Context, localPath, remotePath string,
|
||||
return a.uploadSimple(ctx, file, blobName, fileSize, progress)
|
||||
}
|
||||
|
||||
// uploadSimple uploads a file using simple upload (single request)
|
||||
// uploadSimple uploads a file using simple upload (single request) with retry
|
||||
func (a *AzureBackend) uploadSimple(ctx context.Context, file *os.File, blobName string, fileSize int64, progress ProgressCallback) error {
|
||||
return RetryOperationWithNotify(ctx, DefaultRetryConfig(), func() error {
|
||||
// Reset file position for retry
|
||||
if _, err := file.Seek(0, 0); err != nil {
|
||||
return fmt.Errorf("failed to reset file position: %w", err)
|
||||
}
|
||||
|
||||
blockBlobClient := a.client.ServiceClient().NewContainerClient(a.containerName).NewBlockBlobClient(blobName)
|
||||
|
||||
// Wrap reader with progress tracking
|
||||
@@ -182,6 +188,9 @@ func (a *AzureBackend) uploadSimple(ctx context.Context, file *os.File, blobName
|
||||
}
|
||||
|
||||
return nil
|
||||
}, func(err error, duration time.Duration) {
|
||||
fmt.Printf("[Azure] Upload retry in %v: %v\n", duration, err)
|
||||
})
|
||||
}
|
||||
|
||||
// uploadBlocks uploads a file using block blob staging (for large files)
|
||||
@@ -251,7 +260,7 @@ func (a *AzureBackend) uploadBlocks(ctx context.Context, file *os.File, blobName
|
||||
return nil
|
||||
}
|
||||
|
||||
// Download downloads a file from Azure Blob Storage
|
||||
// Download downloads a file from Azure Blob Storage with retry
|
||||
func (a *AzureBackend) Download(ctx context.Context, remotePath, localPath string, progress ProgressCallback) error {
|
||||
blobName := strings.TrimPrefix(remotePath, "/")
|
||||
blockBlobClient := a.client.ServiceClient().NewContainerClient(a.containerName).NewBlockBlobClient(blobName)
|
||||
@@ -264,6 +273,7 @@ func (a *AzureBackend) Download(ctx context.Context, remotePath, localPath strin
|
||||
|
||||
fileSize := *props.ContentLength
|
||||
|
||||
return RetryOperationWithNotify(ctx, DefaultRetryConfig(), func() error {
|
||||
// Download blob
|
||||
resp, err := blockBlobClient.DownloadStream(ctx, nil)
|
||||
if err != nil {
|
||||
@@ -271,7 +281,7 @@ func (a *AzureBackend) Download(ctx context.Context, remotePath, localPath strin
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
|
||||
// Create local file
|
||||
// Create/truncate local file
|
||||
file, err := os.Create(localPath)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to create file: %w", err)
|
||||
@@ -288,6 +298,9 @@ func (a *AzureBackend) Download(ctx context.Context, remotePath, localPath strin
|
||||
}
|
||||
|
||||
return nil
|
||||
}, func(err error, duration time.Duration) {
|
||||
fmt.Printf("[Azure] Download retry in %v: %v\n", duration, err)
|
||||
})
|
||||
}
|
||||
|
||||
// Delete deletes a file from Azure Blob Storage
|
||||
|
||||
@@ -89,7 +89,7 @@ func (g *GCSBackend) Name() string {
|
||||
return "gcs"
|
||||
}
|
||||
|
||||
// Upload uploads a file to Google Cloud Storage
|
||||
// Upload uploads a file to Google Cloud Storage with retry
|
||||
func (g *GCSBackend) Upload(ctx context.Context, localPath, remotePath string, progress ProgressCallback) error {
|
||||
file, err := os.Open(localPath)
|
||||
if err != nil {
|
||||
@@ -106,6 +106,12 @@ func (g *GCSBackend) Upload(ctx context.Context, localPath, remotePath string, p
|
||||
// Remove leading slash from remote path
|
||||
objectName := strings.TrimPrefix(remotePath, "/")
|
||||
|
||||
return RetryOperationWithNotify(ctx, DefaultRetryConfig(), func() error {
|
||||
// Reset file position for retry
|
||||
if _, err := file.Seek(0, 0); err != nil {
|
||||
return fmt.Errorf("failed to reset file position: %w", err)
|
||||
}
|
||||
|
||||
bucket := g.client.Bucket(g.bucketName)
|
||||
object := bucket.Object(objectName)
|
||||
|
||||
@@ -142,9 +148,12 @@ func (g *GCSBackend) Upload(ctx context.Context, localPath, remotePath string, p
|
||||
}
|
||||
|
||||
return nil
|
||||
}, func(err error, duration time.Duration) {
|
||||
fmt.Printf("[GCS] Upload retry in %v: %v\n", duration, err)
|
||||
})
|
||||
}
|
||||
|
||||
// Download downloads a file from Google Cloud Storage
|
||||
// Download downloads a file from Google Cloud Storage with retry
|
||||
func (g *GCSBackend) Download(ctx context.Context, remotePath, localPath string, progress ProgressCallback) error {
|
||||
objectName := strings.TrimPrefix(remotePath, "/")
|
||||
|
||||
@@ -159,6 +168,7 @@ func (g *GCSBackend) Download(ctx context.Context, remotePath, localPath string,
|
||||
|
||||
fileSize := attrs.Size
|
||||
|
||||
return RetryOperationWithNotify(ctx, DefaultRetryConfig(), func() error {
|
||||
// Create reader
|
||||
reader, err := object.NewReader(ctx)
|
||||
if err != nil {
|
||||
@@ -166,7 +176,7 @@ func (g *GCSBackend) Download(ctx context.Context, remotePath, localPath string,
|
||||
}
|
||||
defer reader.Close()
|
||||
|
||||
// Create local file
|
||||
// Create/truncate local file
|
||||
file, err := os.Create(localPath)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to create file: %w", err)
|
||||
@@ -183,6 +193,9 @@ func (g *GCSBackend) Download(ctx context.Context, remotePath, localPath string,
|
||||
}
|
||||
|
||||
return nil
|
||||
}, func(err error, duration time.Duration) {
|
||||
fmt.Printf("[GCS] Download retry in %v: %v\n", duration, err)
|
||||
})
|
||||
}
|
||||
|
||||
// Delete deletes a file from Google Cloud Storage
|
||||
|
||||
257
internal/cloud/retry.go
Normal file
257
internal/cloud/retry.go
Normal file
@@ -0,0 +1,257 @@
|
||||
package cloud
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"net"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/cenkalti/backoff/v4"
|
||||
)
|
||||
|
||||
// RetryConfig configures retry behavior
|
||||
type RetryConfig struct {
|
||||
MaxRetries int // Maximum number of retries (0 = unlimited)
|
||||
InitialInterval time.Duration // Initial backoff interval
|
||||
MaxInterval time.Duration // Maximum backoff interval
|
||||
MaxElapsedTime time.Duration // Maximum total time for retries
|
||||
Multiplier float64 // Backoff multiplier
|
||||
}
|
||||
|
||||
// DefaultRetryConfig returns sensible defaults for cloud operations
|
||||
func DefaultRetryConfig() *RetryConfig {
|
||||
return &RetryConfig{
|
||||
MaxRetries: 5,
|
||||
InitialInterval: 500 * time.Millisecond,
|
||||
MaxInterval: 30 * time.Second,
|
||||
MaxElapsedTime: 5 * time.Minute,
|
||||
Multiplier: 2.0,
|
||||
}
|
||||
}
|
||||
|
||||
// AggressiveRetryConfig returns config for critical operations that need more retries
|
||||
func AggressiveRetryConfig() *RetryConfig {
|
||||
return &RetryConfig{
|
||||
MaxRetries: 10,
|
||||
InitialInterval: 1 * time.Second,
|
||||
MaxInterval: 60 * time.Second,
|
||||
MaxElapsedTime: 15 * time.Minute,
|
||||
Multiplier: 1.5,
|
||||
}
|
||||
}
|
||||
|
||||
// QuickRetryConfig returns config for operations that should fail fast
|
||||
func QuickRetryConfig() *RetryConfig {
|
||||
return &RetryConfig{
|
||||
MaxRetries: 3,
|
||||
InitialInterval: 100 * time.Millisecond,
|
||||
MaxInterval: 5 * time.Second,
|
||||
MaxElapsedTime: 30 * time.Second,
|
||||
Multiplier: 2.0,
|
||||
}
|
||||
}
|
||||
|
||||
// RetryOperation executes an operation with exponential backoff retry
|
||||
func RetryOperation(ctx context.Context, cfg *RetryConfig, operation func() error) error {
|
||||
if cfg == nil {
|
||||
cfg = DefaultRetryConfig()
|
||||
}
|
||||
|
||||
// Create exponential backoff
|
||||
expBackoff := backoff.NewExponentialBackOff()
|
||||
expBackoff.InitialInterval = cfg.InitialInterval
|
||||
expBackoff.MaxInterval = cfg.MaxInterval
|
||||
expBackoff.MaxElapsedTime = cfg.MaxElapsedTime
|
||||
expBackoff.Multiplier = cfg.Multiplier
|
||||
expBackoff.Reset()
|
||||
|
||||
// Wrap with max retries if specified
|
||||
var b backoff.BackOff = expBackoff
|
||||
if cfg.MaxRetries > 0 {
|
||||
b = backoff.WithMaxRetries(expBackoff, uint64(cfg.MaxRetries))
|
||||
}
|
||||
|
||||
// Add context support
|
||||
b = backoff.WithContext(b, ctx)
|
||||
|
||||
// Track attempts for logging
|
||||
attempt := 0
|
||||
|
||||
// Wrap operation to handle permanent vs retryable errors
|
||||
wrappedOp := func() error {
|
||||
attempt++
|
||||
err := operation()
|
||||
if err == nil {
|
||||
return nil
|
||||
}
|
||||
|
||||
// Check if error is permanent (should not retry)
|
||||
if IsPermanentError(err) {
|
||||
return backoff.Permanent(err)
|
||||
}
|
||||
|
||||
return err
|
||||
}
|
||||
|
||||
return backoff.Retry(wrappedOp, b)
|
||||
}
|
||||
|
||||
// RetryOperationWithNotify executes an operation with retry and calls notify on each retry
|
||||
func RetryOperationWithNotify(ctx context.Context, cfg *RetryConfig, operation func() error, notify func(err error, duration time.Duration)) error {
|
||||
if cfg == nil {
|
||||
cfg = DefaultRetryConfig()
|
||||
}
|
||||
|
||||
// Create exponential backoff
|
||||
expBackoff := backoff.NewExponentialBackOff()
|
||||
expBackoff.InitialInterval = cfg.InitialInterval
|
||||
expBackoff.MaxInterval = cfg.MaxInterval
|
||||
expBackoff.MaxElapsedTime = cfg.MaxElapsedTime
|
||||
expBackoff.Multiplier = cfg.Multiplier
|
||||
expBackoff.Reset()
|
||||
|
||||
// Wrap with max retries if specified
|
||||
var b backoff.BackOff = expBackoff
|
||||
if cfg.MaxRetries > 0 {
|
||||
b = backoff.WithMaxRetries(expBackoff, uint64(cfg.MaxRetries))
|
||||
}
|
||||
|
||||
// Add context support
|
||||
b = backoff.WithContext(b, ctx)
|
||||
|
||||
// Wrap operation to handle permanent vs retryable errors
|
||||
wrappedOp := func() error {
|
||||
err := operation()
|
||||
if err == nil {
|
||||
return nil
|
||||
}
|
||||
|
||||
// Check if error is permanent (should not retry)
|
||||
if IsPermanentError(err) {
|
||||
return backoff.Permanent(err)
|
||||
}
|
||||
|
||||
return err
|
||||
}
|
||||
|
||||
return backoff.RetryNotify(wrappedOp, b, notify)
|
||||
}
|
||||
|
||||
// IsPermanentError returns true if the error should not be retried
|
||||
func IsPermanentError(err error) bool {
|
||||
if err == nil {
|
||||
return false
|
||||
}
|
||||
|
||||
errStr := strings.ToLower(err.Error())
|
||||
|
||||
// Authentication/authorization errors - don't retry
|
||||
permanentPatterns := []string{
|
||||
"access denied",
|
||||
"forbidden",
|
||||
"unauthorized",
|
||||
"invalid credentials",
|
||||
"invalid access key",
|
||||
"invalid secret",
|
||||
"no such bucket",
|
||||
"bucket not found",
|
||||
"container not found",
|
||||
"nosuchbucket",
|
||||
"nosuchkey",
|
||||
"invalid argument",
|
||||
"malformed",
|
||||
"invalid request",
|
||||
"permission denied",
|
||||
"access control",
|
||||
"policy",
|
||||
}
|
||||
|
||||
for _, pattern := range permanentPatterns {
|
||||
if strings.Contains(errStr, pattern) {
|
||||
return true
|
||||
}
|
||||
}
|
||||
|
||||
return false
|
||||
}
|
||||
|
||||
// IsRetryableError returns true if the error is transient and should be retried
|
||||
func IsRetryableError(err error) bool {
|
||||
if err == nil {
|
||||
return false
|
||||
}
|
||||
|
||||
// Network errors are typically retryable
|
||||
var netErr net.Error
|
||||
if ok := isNetError(err, &netErr); ok {
|
||||
return netErr.Timeout() || netErr.Temporary()
|
||||
}
|
||||
|
||||
errStr := strings.ToLower(err.Error())
|
||||
|
||||
// Transient errors - should retry
|
||||
retryablePatterns := []string{
|
||||
"timeout",
|
||||
"connection reset",
|
||||
"connection refused",
|
||||
"connection closed",
|
||||
"eof",
|
||||
"broken pipe",
|
||||
"temporary failure",
|
||||
"service unavailable",
|
||||
"internal server error",
|
||||
"bad gateway",
|
||||
"gateway timeout",
|
||||
"too many requests",
|
||||
"rate limit",
|
||||
"throttl",
|
||||
"slowdown",
|
||||
"try again",
|
||||
"retry",
|
||||
}
|
||||
|
||||
for _, pattern := range retryablePatterns {
|
||||
if strings.Contains(errStr, pattern) {
|
||||
return true
|
||||
}
|
||||
}
|
||||
|
||||
return false
|
||||
}
|
||||
|
||||
// isNetError checks if err wraps a net.Error
|
||||
func isNetError(err error, target *net.Error) bool {
|
||||
for err != nil {
|
||||
if ne, ok := err.(net.Error); ok {
|
||||
*target = ne
|
||||
return true
|
||||
}
|
||||
// Try to unwrap
|
||||
if unwrapper, ok := err.(interface{ Unwrap() error }); ok {
|
||||
err = unwrapper.Unwrap()
|
||||
} else {
|
||||
break
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// WithRetry is a helper that wraps a function with default retry logic
|
||||
func WithRetry(ctx context.Context, operationName string, fn func() error) error {
|
||||
notify := func(err error, duration time.Duration) {
|
||||
// Log retry attempts (caller can provide their own logger if needed)
|
||||
fmt.Printf("[RETRY] %s failed, retrying in %v: %v\n", operationName, duration, err)
|
||||
}
|
||||
|
||||
return RetryOperationWithNotify(ctx, DefaultRetryConfig(), fn, notify)
|
||||
}
|
||||
|
||||
// WithRetryConfig is a helper that wraps a function with custom retry config
|
||||
func WithRetryConfig(ctx context.Context, cfg *RetryConfig, operationName string, fn func() error) error {
|
||||
notify := func(err error, duration time.Duration) {
|
||||
fmt.Printf("[RETRY] %s failed, retrying in %v: %v\n", operationName, duration, err)
|
||||
}
|
||||
|
||||
return RetryOperationWithNotify(ctx, cfg, fn, notify)
|
||||
}
|
||||
@@ -7,6 +7,7 @@ import (
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/aws/aws-sdk-go-v2/aws"
|
||||
"github.com/aws/aws-sdk-go-v2/config"
|
||||
@@ -123,8 +124,14 @@ func (s *S3Backend) Upload(ctx context.Context, localPath, remotePath string, pr
|
||||
return s.uploadSimple(ctx, file, key, fileSize, progress)
|
||||
}
|
||||
|
||||
// uploadSimple performs a simple single-part upload
|
||||
// uploadSimple performs a simple single-part upload with retry
|
||||
func (s *S3Backend) uploadSimple(ctx context.Context, file *os.File, key string, fileSize int64, progress ProgressCallback) error {
|
||||
return RetryOperationWithNotify(ctx, DefaultRetryConfig(), func() error {
|
||||
// Reset file position for retry
|
||||
if _, err := file.Seek(0, 0); err != nil {
|
||||
return fmt.Errorf("failed to reset file position: %w", err)
|
||||
}
|
||||
|
||||
// Create progress reader
|
||||
var reader io.Reader = file
|
||||
if progress != nil {
|
||||
@@ -143,10 +150,19 @@ func (s *S3Backend) uploadSimple(ctx context.Context, file *os.File, key string,
|
||||
}
|
||||
|
||||
return nil
|
||||
}, func(err error, duration time.Duration) {
|
||||
fmt.Printf("[S3] Upload retry in %v: %v\n", duration, err)
|
||||
})
|
||||
}
|
||||
|
||||
// uploadMultipart performs a multipart upload for large files
|
||||
// uploadMultipart performs a multipart upload for large files with retry
|
||||
func (s *S3Backend) uploadMultipart(ctx context.Context, file *os.File, key string, fileSize int64, progress ProgressCallback) error {
|
||||
return RetryOperationWithNotify(ctx, AggressiveRetryConfig(), func() error {
|
||||
// Reset file position for retry
|
||||
if _, err := file.Seek(0, 0); err != nil {
|
||||
return fmt.Errorf("failed to reset file position: %w", err)
|
||||
}
|
||||
|
||||
// Create uploader with custom options
|
||||
uploader := manager.NewUploader(s.client, func(u *manager.Uploader) {
|
||||
// Part size: 10MB
|
||||
@@ -177,9 +193,12 @@ func (s *S3Backend) uploadMultipart(ctx context.Context, file *os.File, key stri
|
||||
}
|
||||
|
||||
return nil
|
||||
}, func(err error, duration time.Duration) {
|
||||
fmt.Printf("[S3] Multipart upload retry in %v: %v\n", duration, err)
|
||||
})
|
||||
}
|
||||
|
||||
// Download downloads a file from S3
|
||||
// Download downloads a file from S3 with retry
|
||||
func (s *S3Backend) Download(ctx context.Context, remotePath, localPath string, progress ProgressCallback) error {
|
||||
// Build S3 key
|
||||
key := s.buildKey(remotePath)
|
||||
@@ -190,6 +209,12 @@ func (s *S3Backend) Download(ctx context.Context, remotePath, localPath string,
|
||||
return fmt.Errorf("failed to get object size: %w", err)
|
||||
}
|
||||
|
||||
// Create directory for local file
|
||||
if err := os.MkdirAll(filepath.Dir(localPath), 0755); err != nil {
|
||||
return fmt.Errorf("failed to create directory: %w", err)
|
||||
}
|
||||
|
||||
return RetryOperationWithNotify(ctx, DefaultRetryConfig(), func() error {
|
||||
// Download from S3
|
||||
result, err := s.client.GetObject(ctx, &s3.GetObjectInput{
|
||||
Bucket: aws.String(s.bucket),
|
||||
@@ -200,11 +225,7 @@ func (s *S3Backend) Download(ctx context.Context, remotePath, localPath string,
|
||||
}
|
||||
defer result.Body.Close()
|
||||
|
||||
// Create local file
|
||||
if err := os.MkdirAll(filepath.Dir(localPath), 0755); err != nil {
|
||||
return fmt.Errorf("failed to create directory: %w", err)
|
||||
}
|
||||
|
||||
// Create/truncate local file
|
||||
outFile, err := os.Create(localPath)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to create local file: %w", err)
|
||||
@@ -223,6 +244,9 @@ func (s *S3Backend) Download(ctx context.Context, remotePath, localPath string,
|
||||
}
|
||||
|
||||
return nil
|
||||
}, func(err error, duration time.Duration) {
|
||||
fmt.Printf("[S3] Download retry in %v: %v\n", duration, err)
|
||||
})
|
||||
}
|
||||
|
||||
// List lists all backup files in S3
|
||||
|
||||
223
internal/fs/fs.go
Normal file
223
internal/fs/fs.go
Normal file
@@ -0,0 +1,223 @@
|
||||
// Package fs provides filesystem abstraction using spf13/afero for testability.
|
||||
// It allows swapping the real filesystem with an in-memory mock for unit tests.
|
||||
package fs
|
||||
|
||||
import (
|
||||
"io"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"time"
|
||||
|
||||
"github.com/spf13/afero"
|
||||
)
|
||||
|
||||
// FS is the global filesystem interface used throughout the application.
|
||||
// By default, it uses the real OS filesystem.
|
||||
// For testing, use SetFS(afero.NewMemMapFs()) to use an in-memory filesystem.
|
||||
var FS afero.Fs = afero.NewOsFs()
|
||||
|
||||
// SetFS sets the global filesystem (useful for testing)
|
||||
func SetFS(fs afero.Fs) {
|
||||
FS = fs
|
||||
}
|
||||
|
||||
// ResetFS resets to the real OS filesystem
|
||||
func ResetFS() {
|
||||
FS = afero.NewOsFs()
|
||||
}
|
||||
|
||||
// NewMemMapFs creates a new in-memory filesystem for testing
|
||||
func NewMemMapFs() afero.Fs {
|
||||
return afero.NewMemMapFs()
|
||||
}
|
||||
|
||||
// NewReadOnlyFs wraps a filesystem to make it read-only
|
||||
func NewReadOnlyFs(base afero.Fs) afero.Fs {
|
||||
return afero.NewReadOnlyFs(base)
|
||||
}
|
||||
|
||||
// NewBasePathFs creates a filesystem rooted at a specific path
|
||||
func NewBasePathFs(base afero.Fs, path string) afero.Fs {
|
||||
return afero.NewBasePathFs(base, path)
|
||||
}
|
||||
|
||||
// --- File Operations (use global FS) ---
|
||||
|
||||
// Create creates a file
|
||||
func Create(name string) (afero.File, error) {
|
||||
return FS.Create(name)
|
||||
}
|
||||
|
||||
// Open opens a file for reading
|
||||
func Open(name string) (afero.File, error) {
|
||||
return FS.Open(name)
|
||||
}
|
||||
|
||||
// OpenFile opens a file with specified flags and permissions
|
||||
func OpenFile(name string, flag int, perm os.FileMode) (afero.File, error) {
|
||||
return FS.OpenFile(name, flag, perm)
|
||||
}
|
||||
|
||||
// Remove removes a file or empty directory
|
||||
func Remove(name string) error {
|
||||
return FS.Remove(name)
|
||||
}
|
||||
|
||||
// RemoveAll removes a path and any children it contains
|
||||
func RemoveAll(path string) error {
|
||||
return FS.RemoveAll(path)
|
||||
}
|
||||
|
||||
// Rename renames (moves) a file
|
||||
func Rename(oldname, newname string) error {
|
||||
return FS.Rename(oldname, newname)
|
||||
}
|
||||
|
||||
// Stat returns file info
|
||||
func Stat(name string) (os.FileInfo, error) {
|
||||
return FS.Stat(name)
|
||||
}
|
||||
|
||||
// Chmod changes file mode
|
||||
func Chmod(name string, mode os.FileMode) error {
|
||||
return FS.Chmod(name, mode)
|
||||
}
|
||||
|
||||
// Chown changes file ownership (may not work on all filesystems)
|
||||
func Chown(name string, uid, gid int) error {
|
||||
return FS.Chown(name, uid, gid)
|
||||
}
|
||||
|
||||
// Chtimes changes file access and modification times
|
||||
func Chtimes(name string, atime, mtime time.Time) error {
|
||||
return FS.Chtimes(name, atime, mtime)
|
||||
}
|
||||
|
||||
// --- Directory Operations ---
|
||||
|
||||
// Mkdir creates a directory
|
||||
func Mkdir(name string, perm os.FileMode) error {
|
||||
return FS.Mkdir(name, perm)
|
||||
}
|
||||
|
||||
// MkdirAll creates a directory and all parents
|
||||
func MkdirAll(path string, perm os.FileMode) error {
|
||||
return FS.MkdirAll(path, perm)
|
||||
}
|
||||
|
||||
// ReadDir reads a directory
|
||||
func ReadDir(dirname string) ([]os.FileInfo, error) {
|
||||
return afero.ReadDir(FS, dirname)
|
||||
}
|
||||
|
||||
// --- File Content Operations ---
|
||||
|
||||
// ReadFile reads an entire file
|
||||
func ReadFile(filename string) ([]byte, error) {
|
||||
return afero.ReadFile(FS, filename)
|
||||
}
|
||||
|
||||
// WriteFile writes data to a file
|
||||
func WriteFile(filename string, data []byte, perm os.FileMode) error {
|
||||
return afero.WriteFile(FS, filename, data, perm)
|
||||
}
|
||||
|
||||
// --- Existence Checks ---
|
||||
|
||||
// Exists checks if a file or directory exists
|
||||
func Exists(path string) (bool, error) {
|
||||
return afero.Exists(FS, path)
|
||||
}
|
||||
|
||||
// DirExists checks if a directory exists
|
||||
func DirExists(path string) (bool, error) {
|
||||
return afero.DirExists(FS, path)
|
||||
}
|
||||
|
||||
// IsDir checks if path is a directory
|
||||
func IsDir(path string) (bool, error) {
|
||||
return afero.IsDir(FS, path)
|
||||
}
|
||||
|
||||
// IsEmpty checks if a directory is empty
|
||||
func IsEmpty(path string) (bool, error) {
|
||||
return afero.IsEmpty(FS, path)
|
||||
}
|
||||
|
||||
// --- Utility Functions ---
|
||||
|
||||
// Walk walks a directory tree
|
||||
func Walk(root string, walkFn filepath.WalkFunc) error {
|
||||
return afero.Walk(FS, root, walkFn)
|
||||
}
|
||||
|
||||
// Glob returns the names of all files matching pattern
|
||||
func Glob(pattern string) ([]string, error) {
|
||||
return afero.Glob(FS, pattern)
|
||||
}
|
||||
|
||||
// TempDir creates a temporary directory
|
||||
func TempDir(dir, prefix string) (string, error) {
|
||||
return afero.TempDir(FS, dir, prefix)
|
||||
}
|
||||
|
||||
// TempFile creates a temporary file
|
||||
func TempFile(dir, pattern string) (afero.File, error) {
|
||||
return afero.TempFile(FS, dir, pattern)
|
||||
}
|
||||
|
||||
// CopyFile copies a file from src to dst
|
||||
func CopyFile(src, dst string) error {
|
||||
srcFile, err := FS.Open(src)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer srcFile.Close()
|
||||
|
||||
srcInfo, err := srcFile.Stat()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
dstFile, err := FS.OpenFile(dst, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, srcInfo.Mode())
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer dstFile.Close()
|
||||
|
||||
_, err = io.Copy(dstFile, srcFile)
|
||||
return err
|
||||
}
|
||||
|
||||
// FileSize returns the size of a file
|
||||
func FileSize(path string) (int64, error) {
|
||||
info, err := FS.Stat(path)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
return info.Size(), nil
|
||||
}
|
||||
|
||||
// --- Testing Helpers ---
|
||||
|
||||
// WithMemFs executes a function with an in-memory filesystem, then restores the original
|
||||
func WithMemFs(fn func(fs afero.Fs)) {
|
||||
original := FS
|
||||
memFs := afero.NewMemMapFs()
|
||||
FS = memFs
|
||||
defer func() { FS = original }()
|
||||
fn(memFs)
|
||||
}
|
||||
|
||||
// SetupTestDir creates a test directory structure in-memory
|
||||
func SetupTestDir(files map[string]string) afero.Fs {
|
||||
memFs := afero.NewMemMapFs()
|
||||
for path, content := range files {
|
||||
dir := filepath.Dir(path)
|
||||
if dir != "." && dir != "/" {
|
||||
_ = memFs.MkdirAll(dir, 0755)
|
||||
}
|
||||
_ = afero.WriteFile(memFs, path, []byte(content), 0644)
|
||||
}
|
||||
return memFs
|
||||
}
|
||||
191
internal/fs/fs_test.go
Normal file
191
internal/fs/fs_test.go
Normal file
@@ -0,0 +1,191 @@
|
||||
package fs
|
||||
|
||||
import (
|
||||
"os"
|
||||
"testing"
|
||||
|
||||
"github.com/spf13/afero"
|
||||
)
|
||||
|
||||
func TestMemMapFs(t *testing.T) {
|
||||
// Use in-memory filesystem for testing
|
||||
WithMemFs(func(memFs afero.Fs) {
|
||||
// Create a file
|
||||
err := WriteFile("/test/file.txt", []byte("hello world"), 0644)
|
||||
if err != nil {
|
||||
t.Fatalf("WriteFile failed: %v", err)
|
||||
}
|
||||
|
||||
// Read it back
|
||||
content, err := ReadFile("/test/file.txt")
|
||||
if err != nil {
|
||||
t.Fatalf("ReadFile failed: %v", err)
|
||||
}
|
||||
|
||||
if string(content) != "hello world" {
|
||||
t.Errorf("expected 'hello world', got '%s'", string(content))
|
||||
}
|
||||
|
||||
// Check existence
|
||||
exists, err := Exists("/test/file.txt")
|
||||
if err != nil {
|
||||
t.Fatalf("Exists failed: %v", err)
|
||||
}
|
||||
if !exists {
|
||||
t.Error("file should exist")
|
||||
}
|
||||
|
||||
// Check non-existent file
|
||||
exists, err = Exists("/nonexistent.txt")
|
||||
if err != nil {
|
||||
t.Fatalf("Exists failed: %v", err)
|
||||
}
|
||||
if exists {
|
||||
t.Error("file should not exist")
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
func TestSetupTestDir(t *testing.T) {
|
||||
// Create test directory structure
|
||||
testFs := SetupTestDir(map[string]string{
|
||||
"/backups/db1.dump": "database 1 content",
|
||||
"/backups/db2.dump": "database 2 content",
|
||||
"/config/settings.json": `{"key": "value"}`,
|
||||
})
|
||||
|
||||
// Verify files exist
|
||||
content, err := afero.ReadFile(testFs, "/backups/db1.dump")
|
||||
if err != nil {
|
||||
t.Fatalf("ReadFile failed: %v", err)
|
||||
}
|
||||
if string(content) != "database 1 content" {
|
||||
t.Errorf("unexpected content: %s", string(content))
|
||||
}
|
||||
|
||||
// Verify directory structure
|
||||
files, err := afero.ReadDir(testFs, "/backups")
|
||||
if err != nil {
|
||||
t.Fatalf("ReadDir failed: %v", err)
|
||||
}
|
||||
if len(files) != 2 {
|
||||
t.Errorf("expected 2 files, got %d", len(files))
|
||||
}
|
||||
}
|
||||
|
||||
func TestCopyFile(t *testing.T) {
|
||||
WithMemFs(func(memFs afero.Fs) {
|
||||
// Create source file
|
||||
err := WriteFile("/source.txt", []byte("copy me"), 0644)
|
||||
if err != nil {
|
||||
t.Fatalf("WriteFile failed: %v", err)
|
||||
}
|
||||
|
||||
// Copy file
|
||||
err = CopyFile("/source.txt", "/dest.txt")
|
||||
if err != nil {
|
||||
t.Fatalf("CopyFile failed: %v", err)
|
||||
}
|
||||
|
||||
// Verify copy
|
||||
content, err := ReadFile("/dest.txt")
|
||||
if err != nil {
|
||||
t.Fatalf("ReadFile failed: %v", err)
|
||||
}
|
||||
if string(content) != "copy me" {
|
||||
t.Errorf("unexpected content: %s", string(content))
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
func TestFileSize(t *testing.T) {
|
||||
WithMemFs(func(memFs afero.Fs) {
|
||||
data := []byte("12345678901234567890") // 20 bytes
|
||||
err := WriteFile("/sized.txt", data, 0644)
|
||||
if err != nil {
|
||||
t.Fatalf("WriteFile failed: %v", err)
|
||||
}
|
||||
|
||||
size, err := FileSize("/sized.txt")
|
||||
if err != nil {
|
||||
t.Fatalf("FileSize failed: %v", err)
|
||||
}
|
||||
if size != 20 {
|
||||
t.Errorf("expected size 20, got %d", size)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
func TestTempDir(t *testing.T) {
|
||||
WithMemFs(func(memFs afero.Fs) {
|
||||
// Create temp dir
|
||||
dir, err := TempDir("", "test-")
|
||||
if err != nil {
|
||||
t.Fatalf("TempDir failed: %v", err)
|
||||
}
|
||||
|
||||
// Verify it exists
|
||||
isDir, err := IsDir(dir)
|
||||
if err != nil {
|
||||
t.Fatalf("IsDir failed: %v", err)
|
||||
}
|
||||
if !isDir {
|
||||
t.Error("temp dir should be a directory")
|
||||
}
|
||||
|
||||
// Verify it's empty
|
||||
isEmpty, err := IsEmpty(dir)
|
||||
if err != nil {
|
||||
t.Fatalf("IsEmpty failed: %v", err)
|
||||
}
|
||||
if !isEmpty {
|
||||
t.Error("temp dir should be empty")
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
func TestWalk(t *testing.T) {
|
||||
WithMemFs(func(memFs afero.Fs) {
|
||||
// Create directory structure
|
||||
_ = MkdirAll("/root/a/b", 0755)
|
||||
_ = WriteFile("/root/file1.txt", []byte("1"), 0644)
|
||||
_ = WriteFile("/root/a/file2.txt", []byte("2"), 0644)
|
||||
_ = WriteFile("/root/a/b/file3.txt", []byte("3"), 0644)
|
||||
|
||||
var files []string
|
||||
err := Walk("/root", func(path string, info os.FileInfo, err error) error {
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if !info.IsDir() {
|
||||
files = append(files, path)
|
||||
}
|
||||
return nil
|
||||
})
|
||||
|
||||
if err != nil {
|
||||
t.Fatalf("Walk failed: %v", err)
|
||||
}
|
||||
|
||||
if len(files) != 3 {
|
||||
t.Errorf("expected 3 files, got %d: %v", len(files), files)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
func TestGlob(t *testing.T) {
|
||||
WithMemFs(func(memFs afero.Fs) {
|
||||
_ = WriteFile("/data/backup1.dump", []byte("1"), 0644)
|
||||
_ = WriteFile("/data/backup2.dump", []byte("2"), 0644)
|
||||
_ = WriteFile("/data/config.json", []byte("{}"), 0644)
|
||||
|
||||
matches, err := Glob("/data/*.dump")
|
||||
if err != nil {
|
||||
t.Fatalf("Glob failed: %v", err)
|
||||
}
|
||||
|
||||
if len(matches) != 2 {
|
||||
t.Errorf("expected 2 matches, got %d: %v", len(matches), matches)
|
||||
}
|
||||
})
|
||||
}
|
||||
@@ -368,7 +368,7 @@ func (d *Diagnoser) diagnoseSQLScript(filePath string, compressed bool, result *
|
||||
}
|
||||
|
||||
// Store last line for termination check
|
||||
if lineNumber > 0 && (lineNumber%100000 == 0) && d.verbose {
|
||||
if lineNumber > 0 && (lineNumber%100000 == 0) && d.verbose && d.log != nil {
|
||||
d.log.Debug("Scanning SQL file", "lines_processed", lineNumber)
|
||||
}
|
||||
}
|
||||
@@ -430,9 +430,11 @@ func (d *Diagnoser) diagnoseClusterArchive(filePath string, result *DiagnoseResu
|
||||
}
|
||||
}
|
||||
|
||||
if d.log != nil {
|
||||
d.log.Info("Verifying cluster archive integrity",
|
||||
"size", fmt.Sprintf("%.1f GB", float64(result.FileSize)/(1024*1024*1024)),
|
||||
"timeout", fmt.Sprintf("%d min", timeoutMinutes))
|
||||
}
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), time.Duration(timeoutMinutes)*time.Minute)
|
||||
defer cancel()
|
||||
@@ -561,7 +563,7 @@ func (d *Diagnoser) diagnoseClusterArchive(filePath string, result *DiagnoseResu
|
||||
}
|
||||
|
||||
// For verbose mode, diagnose individual dumps inside the archive
|
||||
if d.verbose && len(dumpFiles) > 0 {
|
||||
if d.verbose && len(dumpFiles) > 0 && d.log != nil {
|
||||
d.log.Info("Cluster archive contains databases", "count", len(dumpFiles))
|
||||
for _, df := range dumpFiles {
|
||||
d.log.Info(" - " + df)
|
||||
@@ -684,9 +686,11 @@ func (d *Diagnoser) DiagnoseClusterDumps(archivePath, tempDir string) ([]*Diagno
|
||||
}
|
||||
}
|
||||
|
||||
if d.log != nil {
|
||||
d.log.Info("Listing cluster archive contents",
|
||||
"size", fmt.Sprintf("%.1f GB", float64(archiveInfo.Size())/(1024*1024*1024)),
|
||||
"timeout", fmt.Sprintf("%d min", timeoutMinutes))
|
||||
}
|
||||
|
||||
listCtx, listCancel := context.WithTimeout(context.Background(), time.Duration(timeoutMinutes)*time.Minute)
|
||||
defer listCancel()
|
||||
@@ -766,7 +770,9 @@ func (d *Diagnoser) DiagnoseClusterDumps(archivePath, tempDir string) ([]*Diagno
|
||||
return []*DiagnoseResult{errResult}, nil
|
||||
}
|
||||
|
||||
if d.log != nil {
|
||||
d.log.Debug("Archive listing streamed successfully", "total_files", fileCount, "relevant_files", len(files))
|
||||
}
|
||||
|
||||
// Check if we have enough disk space (estimate 4x archive size needed)
|
||||
// archiveInfo already obtained at function start
|
||||
@@ -781,7 +787,9 @@ func (d *Diagnoser) DiagnoseClusterDumps(archivePath, tempDir string) ([]*Diagno
|
||||
testCancel()
|
||||
}
|
||||
|
||||
if d.log != nil {
|
||||
d.log.Info("Archive listing successful", "files", len(files))
|
||||
}
|
||||
|
||||
// Try full extraction - NO TIMEOUT here as large archives can take a long time
|
||||
// Use a generous timeout (30 minutes) for very large archives
|
||||
@@ -870,11 +878,15 @@ func (d *Diagnoser) DiagnoseClusterDumps(archivePath, tempDir string) ([]*Diagno
|
||||
}
|
||||
|
||||
dumpPath := filepath.Join(dumpsDir, name)
|
||||
if d.log != nil {
|
||||
d.log.Info("Diagnosing dump file", "file", name)
|
||||
}
|
||||
|
||||
result, err := d.DiagnoseFile(dumpPath)
|
||||
if err != nil {
|
||||
if d.log != nil {
|
||||
d.log.Warn("Failed to diagnose file", "file", name, "error", err)
|
||||
}
|
||||
continue
|
||||
}
|
||||
results = append(results, result)
|
||||
|
||||
@@ -1,9 +1,12 @@
|
||||
package restore
|
||||
|
||||
import (
|
||||
"archive/tar"
|
||||
"compress/gzip"
|
||||
"context"
|
||||
"database/sql"
|
||||
"fmt"
|
||||
"io"
|
||||
"os"
|
||||
"os/exec"
|
||||
"path/filepath"
|
||||
@@ -24,6 +27,13 @@ import (
|
||||
_ "github.com/jackc/pgx/v5/stdlib" // PostgreSQL driver
|
||||
)
|
||||
|
||||
// ProgressCallback is called with progress updates during long operations
|
||||
// Parameters: current bytes/items done, total bytes/items, description
|
||||
type ProgressCallback func(current, total int64, description string)
|
||||
|
||||
// DatabaseProgressCallback is called with database count progress during cluster restore
|
||||
type DatabaseProgressCallback func(done, total int, dbName string)
|
||||
|
||||
// Engine handles database restore operations
|
||||
type Engine struct {
|
||||
cfg *config.Config
|
||||
@@ -33,6 +43,10 @@ type Engine struct {
|
||||
detailedReporter *progress.DetailedReporter
|
||||
dryRun bool
|
||||
debugLogPath string // Path to save debug log on error
|
||||
|
||||
// TUI progress callback for detailed progress reporting
|
||||
progressCallback ProgressCallback
|
||||
dbProgressCallback DatabaseProgressCallback
|
||||
}
|
||||
|
||||
// New creates a new restore engine
|
||||
@@ -88,6 +102,30 @@ func (e *Engine) SetDebugLogPath(path string) {
|
||||
e.debugLogPath = path
|
||||
}
|
||||
|
||||
// SetProgressCallback sets a callback for detailed progress reporting (for TUI mode)
|
||||
func (e *Engine) SetProgressCallback(cb ProgressCallback) {
|
||||
e.progressCallback = cb
|
||||
}
|
||||
|
||||
// SetDatabaseProgressCallback sets a callback for database count progress during cluster restore
|
||||
func (e *Engine) SetDatabaseProgressCallback(cb DatabaseProgressCallback) {
|
||||
e.dbProgressCallback = cb
|
||||
}
|
||||
|
||||
// reportProgress safely calls the progress callback if set
|
||||
func (e *Engine) reportProgress(current, total int64, description string) {
|
||||
if e.progressCallback != nil {
|
||||
e.progressCallback(current, total, description)
|
||||
}
|
||||
}
|
||||
|
||||
// reportDatabaseProgress safely calls the database progress callback if set
|
||||
func (e *Engine) reportDatabaseProgress(done, total int, dbName string) {
|
||||
if e.dbProgressCallback != nil {
|
||||
e.dbProgressCallback(done, total, dbName)
|
||||
}
|
||||
}
|
||||
|
||||
// loggerAdapter adapts our logger to the progress.Logger interface
|
||||
type loggerAdapter struct {
|
||||
logger logger.Logger
|
||||
@@ -1040,6 +1078,8 @@ func (e *Engine) RestoreCluster(ctx context.Context, archivePath string) error {
|
||||
statusMsg := fmt.Sprintf("Restoring database %s (%d/%d)", dbName, idx+1, totalDBs)
|
||||
e.progress.Update(statusMsg)
|
||||
e.log.Info("Restoring database", "name", dbName, "file", dumpFile, "progress", dbProgress)
|
||||
// Report database progress for TUI
|
||||
e.reportDatabaseProgress(idx, totalDBs, dbName)
|
||||
mu.Unlock()
|
||||
|
||||
// STEP 1: Drop existing database completely (clean slate)
|
||||
@@ -1146,8 +1186,144 @@ func (e *Engine) RestoreCluster(ctx context.Context, archivePath string) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// extractArchive extracts a tar.gz archive
|
||||
// extractArchive extracts a tar.gz archive with progress reporting
|
||||
func (e *Engine) extractArchive(ctx context.Context, archivePath, destDir string) error {
|
||||
// If progress callback is set, use Go's archive/tar for progress tracking
|
||||
if e.progressCallback != nil {
|
||||
return e.extractArchiveWithProgress(ctx, archivePath, destDir)
|
||||
}
|
||||
|
||||
// Otherwise use fast shell tar (no progress)
|
||||
return e.extractArchiveShell(ctx, archivePath, destDir)
|
||||
}
|
||||
|
||||
// extractArchiveWithProgress extracts using Go's archive/tar with detailed progress reporting
|
||||
func (e *Engine) extractArchiveWithProgress(ctx context.Context, archivePath, destDir string) error {
|
||||
// Get archive size for progress calculation
|
||||
archiveInfo, err := os.Stat(archivePath)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to stat archive: %w", err)
|
||||
}
|
||||
totalSize := archiveInfo.Size()
|
||||
|
||||
// Open the archive file
|
||||
file, err := os.Open(archivePath)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to open archive: %w", err)
|
||||
}
|
||||
defer file.Close()
|
||||
|
||||
// Wrap with progress reader
|
||||
progressReader := &progressReader{
|
||||
reader: file,
|
||||
totalSize: totalSize,
|
||||
callback: e.progressCallback,
|
||||
desc: "Extracting archive",
|
||||
}
|
||||
|
||||
// Create gzip reader
|
||||
gzReader, err := gzip.NewReader(progressReader)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to create gzip reader: %w", err)
|
||||
}
|
||||
defer gzReader.Close()
|
||||
|
||||
// Create tar reader
|
||||
tarReader := tar.NewReader(gzReader)
|
||||
|
||||
// Extract files
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
default:
|
||||
}
|
||||
|
||||
header, err := tarReader.Next()
|
||||
if err == io.EOF {
|
||||
break // End of archive
|
||||
}
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to read tar header: %w", err)
|
||||
}
|
||||
|
||||
// Sanitize and validate path
|
||||
targetPath := filepath.Join(destDir, header.Name)
|
||||
|
||||
// Security check: ensure path is within destDir (prevent path traversal)
|
||||
if !strings.HasPrefix(filepath.Clean(targetPath), filepath.Clean(destDir)) {
|
||||
e.log.Warn("Skipping potentially malicious path in archive", "path", header.Name)
|
||||
continue
|
||||
}
|
||||
|
||||
switch header.Typeflag {
|
||||
case tar.TypeDir:
|
||||
if err := os.MkdirAll(targetPath, 0755); err != nil {
|
||||
return fmt.Errorf("failed to create directory %s: %w", targetPath, err)
|
||||
}
|
||||
case tar.TypeReg:
|
||||
// Ensure parent directory exists
|
||||
if err := os.MkdirAll(filepath.Dir(targetPath), 0755); err != nil {
|
||||
return fmt.Errorf("failed to create parent directory: %w", err)
|
||||
}
|
||||
|
||||
// Create the file
|
||||
outFile, err := os.OpenFile(targetPath, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, os.FileMode(header.Mode))
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to create file %s: %w", targetPath, err)
|
||||
}
|
||||
|
||||
// Copy file contents
|
||||
if _, err := io.Copy(outFile, tarReader); err != nil {
|
||||
outFile.Close()
|
||||
return fmt.Errorf("failed to write file %s: %w", targetPath, err)
|
||||
}
|
||||
outFile.Close()
|
||||
case tar.TypeSymlink:
|
||||
// Handle symlinks (common in some archives)
|
||||
if err := os.Symlink(header.Linkname, targetPath); err != nil {
|
||||
// Ignore symlink errors (may already exist or not supported)
|
||||
e.log.Debug("Could not create symlink", "path", targetPath, "target", header.Linkname)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Final progress update
|
||||
e.reportProgress(totalSize, totalSize, "Extraction complete")
|
||||
return nil
|
||||
}
|
||||
|
||||
// progressReader wraps an io.Reader to report read progress
|
||||
type progressReader struct {
|
||||
reader io.Reader
|
||||
totalSize int64
|
||||
bytesRead int64
|
||||
callback ProgressCallback
|
||||
desc string
|
||||
lastReport time.Time
|
||||
reportEvery time.Duration
|
||||
}
|
||||
|
||||
func (pr *progressReader) Read(p []byte) (n int, err error) {
|
||||
n, err = pr.reader.Read(p)
|
||||
pr.bytesRead += int64(n)
|
||||
|
||||
// Throttle progress reporting to every 100ms
|
||||
if pr.reportEvery == 0 {
|
||||
pr.reportEvery = 100 * time.Millisecond
|
||||
}
|
||||
if time.Since(pr.lastReport) > pr.reportEvery {
|
||||
if pr.callback != nil {
|
||||
pr.callback(pr.bytesRead, pr.totalSize, pr.desc)
|
||||
}
|
||||
pr.lastReport = time.Now()
|
||||
}
|
||||
|
||||
return n, err
|
||||
}
|
||||
|
||||
// extractArchiveShell extracts using shell tar command (faster but no progress)
|
||||
func (e *Engine) extractArchiveShell(ctx context.Context, archivePath, destDir string) error {
|
||||
cmd := exec.CommandContext(ctx, "tar", "-xzf", archivePath, "-C", destDir)
|
||||
|
||||
// Stream stderr to avoid memory issues - tar can produce lots of output for large archives
|
||||
|
||||
@@ -255,7 +255,9 @@ func (s *Safety) CheckDiskSpaceAt(archivePath string, checkDir string, multiplie
|
||||
// Get available disk space
|
||||
availableSpace, err := getDiskSpace(checkDir)
|
||||
if err != nil {
|
||||
if s.log != nil {
|
||||
s.log.Warn("Cannot check disk space", "error", err)
|
||||
}
|
||||
return nil // Don't fail if we can't check
|
||||
}
|
||||
|
||||
@@ -278,10 +280,12 @@ func (s *Safety) CheckDiskSpaceAt(archivePath string, checkDir string, multiplie
|
||||
checkDir)
|
||||
}
|
||||
|
||||
if s.log != nil {
|
||||
s.log.Info("Disk space check passed",
|
||||
"location", checkDir,
|
||||
"required", FormatBytes(requiredSpace),
|
||||
"available", FormatBytes(availableSpace))
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -251,13 +251,13 @@ func (m ArchiveBrowserModel) View() string {
|
||||
var s strings.Builder
|
||||
|
||||
// Header
|
||||
title := "[PKG] Backup Archives"
|
||||
title := "[SELECT] Backup Archives"
|
||||
if m.mode == "restore-single" {
|
||||
title = "[PKG] Select Archive to Restore (Single Database)"
|
||||
title = "[SELECT] Select Archive to Restore (Single Database)"
|
||||
} else if m.mode == "restore-cluster" {
|
||||
title = "[PKG] Select Archive to Restore (Cluster)"
|
||||
title = "[SELECT] Select Archive to Restore (Cluster)"
|
||||
} else if m.mode == "diagnose" {
|
||||
title = "[SEARCH] Select Archive to Diagnose"
|
||||
title = "[SELECT] Select Archive to Diagnose"
|
||||
}
|
||||
|
||||
s.WriteString(titleStyle.Render(title))
|
||||
|
||||
@@ -230,7 +230,7 @@ func (m BackupManagerModel) View() string {
|
||||
var s strings.Builder
|
||||
|
||||
// Title
|
||||
s.WriteString(TitleStyle.Render("[DB] Backup Archive Manager"))
|
||||
s.WriteString(TitleStyle.Render("[SELECT] Backup Archive Manager"))
|
||||
s.WriteString("\n\n")
|
||||
|
||||
// Status line (no box, bold+color accents)
|
||||
|
||||
406
internal/tui/detailed_progress.go
Normal file
406
internal/tui/detailed_progress.go
Normal file
@@ -0,0 +1,406 @@
|
||||
package tui
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
// DetailedProgress provides schollz-like progress information for TUI rendering
|
||||
// This is a data structure that can be queried by Bubble Tea's View() method
|
||||
type DetailedProgress struct {
|
||||
mu sync.RWMutex
|
||||
|
||||
// Core progress
|
||||
Total int64 // Total bytes or items
|
||||
Current int64 // Current bytes or items done
|
||||
|
||||
// Display info
|
||||
Description string // What operation is happening
|
||||
Unit string // "bytes", "files", "databases", etc.
|
||||
|
||||
// Timing for ETA/speed calculation
|
||||
StartTime time.Time
|
||||
LastUpdate time.Time
|
||||
SpeedWindow []speedSample // Rolling window for speed calculation
|
||||
|
||||
// State
|
||||
IsIndeterminate bool // True if total is unknown (spinner mode)
|
||||
IsComplete bool
|
||||
IsFailed bool
|
||||
ErrorMessage string
|
||||
}
|
||||
|
||||
type speedSample struct {
|
||||
timestamp time.Time
|
||||
bytes int64
|
||||
}
|
||||
|
||||
// NewDetailedProgress creates a progress tracker with known total
|
||||
func NewDetailedProgress(total int64, description string) *DetailedProgress {
|
||||
return &DetailedProgress{
|
||||
Total: total,
|
||||
Description: description,
|
||||
Unit: "bytes",
|
||||
StartTime: time.Now(),
|
||||
LastUpdate: time.Now(),
|
||||
SpeedWindow: make([]speedSample, 0, 20),
|
||||
IsIndeterminate: total <= 0,
|
||||
}
|
||||
}
|
||||
|
||||
// NewDetailedProgressItems creates a progress tracker for item counts
|
||||
func NewDetailedProgressItems(total int, description string) *DetailedProgress {
|
||||
return &DetailedProgress{
|
||||
Total: int64(total),
|
||||
Description: description,
|
||||
Unit: "items",
|
||||
StartTime: time.Now(),
|
||||
LastUpdate: time.Now(),
|
||||
SpeedWindow: make([]speedSample, 0, 20),
|
||||
IsIndeterminate: total <= 0,
|
||||
}
|
||||
}
|
||||
|
||||
// NewDetailedProgressSpinner creates an indeterminate progress tracker
|
||||
func NewDetailedProgressSpinner(description string) *DetailedProgress {
|
||||
return &DetailedProgress{
|
||||
Total: -1,
|
||||
Description: description,
|
||||
Unit: "",
|
||||
StartTime: time.Now(),
|
||||
LastUpdate: time.Now(),
|
||||
SpeedWindow: make([]speedSample, 0, 20),
|
||||
IsIndeterminate: true,
|
||||
}
|
||||
}
|
||||
|
||||
// Add adds to the current progress
|
||||
func (dp *DetailedProgress) Add(n int64) {
|
||||
dp.mu.Lock()
|
||||
defer dp.mu.Unlock()
|
||||
|
||||
dp.Current += n
|
||||
dp.LastUpdate = time.Now()
|
||||
|
||||
// Add speed sample
|
||||
dp.SpeedWindow = append(dp.SpeedWindow, speedSample{
|
||||
timestamp: dp.LastUpdate,
|
||||
bytes: dp.Current,
|
||||
})
|
||||
|
||||
// Keep only last 20 samples for speed calculation
|
||||
if len(dp.SpeedWindow) > 20 {
|
||||
dp.SpeedWindow = dp.SpeedWindow[len(dp.SpeedWindow)-20:]
|
||||
}
|
||||
}
|
||||
|
||||
// Set sets the current progress to a specific value
|
||||
func (dp *DetailedProgress) Set(n int64) {
|
||||
dp.mu.Lock()
|
||||
defer dp.mu.Unlock()
|
||||
|
||||
dp.Current = n
|
||||
dp.LastUpdate = time.Now()
|
||||
|
||||
// Add speed sample
|
||||
dp.SpeedWindow = append(dp.SpeedWindow, speedSample{
|
||||
timestamp: dp.LastUpdate,
|
||||
bytes: dp.Current,
|
||||
})
|
||||
|
||||
if len(dp.SpeedWindow) > 20 {
|
||||
dp.SpeedWindow = dp.SpeedWindow[len(dp.SpeedWindow)-20:]
|
||||
}
|
||||
}
|
||||
|
||||
// SetTotal updates the total (useful when total becomes known during operation)
|
||||
func (dp *DetailedProgress) SetTotal(total int64) {
|
||||
dp.mu.Lock()
|
||||
defer dp.mu.Unlock()
|
||||
|
||||
dp.Total = total
|
||||
dp.IsIndeterminate = total <= 0
|
||||
}
|
||||
|
||||
// SetDescription updates the description
|
||||
func (dp *DetailedProgress) SetDescription(desc string) {
|
||||
dp.mu.Lock()
|
||||
defer dp.mu.Unlock()
|
||||
dp.Description = desc
|
||||
}
|
||||
|
||||
// Complete marks the progress as complete
|
||||
func (dp *DetailedProgress) Complete() {
|
||||
dp.mu.Lock()
|
||||
defer dp.mu.Unlock()
|
||||
|
||||
dp.IsComplete = true
|
||||
dp.Current = dp.Total
|
||||
}
|
||||
|
||||
// Fail marks the progress as failed
|
||||
func (dp *DetailedProgress) Fail(errMsg string) {
|
||||
dp.mu.Lock()
|
||||
defer dp.mu.Unlock()
|
||||
|
||||
dp.IsFailed = true
|
||||
dp.ErrorMessage = errMsg
|
||||
}
|
||||
|
||||
// GetPercent returns the progress percentage (0-100)
|
||||
func (dp *DetailedProgress) GetPercent() int {
|
||||
dp.mu.RLock()
|
||||
defer dp.mu.RUnlock()
|
||||
|
||||
if dp.IsIndeterminate || dp.Total <= 0 {
|
||||
return 0
|
||||
}
|
||||
percent := int((dp.Current * 100) / dp.Total)
|
||||
if percent > 100 {
|
||||
return 100
|
||||
}
|
||||
return percent
|
||||
}
|
||||
|
||||
// GetSpeed returns the current transfer speed in bytes/second
|
||||
func (dp *DetailedProgress) GetSpeed() float64 {
|
||||
dp.mu.RLock()
|
||||
defer dp.mu.RUnlock()
|
||||
|
||||
if len(dp.SpeedWindow) < 2 {
|
||||
return 0
|
||||
}
|
||||
|
||||
// Use first and last samples in window for smoothed speed
|
||||
first := dp.SpeedWindow[0]
|
||||
last := dp.SpeedWindow[len(dp.SpeedWindow)-1]
|
||||
|
||||
elapsed := last.timestamp.Sub(first.timestamp).Seconds()
|
||||
if elapsed <= 0 {
|
||||
return 0
|
||||
}
|
||||
|
||||
bytesTransferred := last.bytes - first.bytes
|
||||
return float64(bytesTransferred) / elapsed
|
||||
}
|
||||
|
||||
// GetETA returns the estimated time remaining
|
||||
func (dp *DetailedProgress) GetETA() time.Duration {
|
||||
dp.mu.RLock()
|
||||
defer dp.mu.RUnlock()
|
||||
|
||||
if dp.IsIndeterminate || dp.Total <= 0 || dp.Current >= dp.Total {
|
||||
return 0
|
||||
}
|
||||
|
||||
speed := dp.getSpeedLocked()
|
||||
if speed <= 0 {
|
||||
return 0
|
||||
}
|
||||
|
||||
remaining := dp.Total - dp.Current
|
||||
seconds := float64(remaining) / speed
|
||||
return time.Duration(seconds) * time.Second
|
||||
}
|
||||
|
||||
func (dp *DetailedProgress) getSpeedLocked() float64 {
|
||||
if len(dp.SpeedWindow) < 2 {
|
||||
return 0
|
||||
}
|
||||
|
||||
first := dp.SpeedWindow[0]
|
||||
last := dp.SpeedWindow[len(dp.SpeedWindow)-1]
|
||||
|
||||
elapsed := last.timestamp.Sub(first.timestamp).Seconds()
|
||||
if elapsed <= 0 {
|
||||
return 0
|
||||
}
|
||||
|
||||
bytesTransferred := last.bytes - first.bytes
|
||||
return float64(bytesTransferred) / elapsed
|
||||
}
|
||||
|
||||
// GetElapsed returns the elapsed time since start
|
||||
func (dp *DetailedProgress) GetElapsed() time.Duration {
|
||||
dp.mu.RLock()
|
||||
defer dp.mu.RUnlock()
|
||||
return time.Since(dp.StartTime)
|
||||
}
|
||||
|
||||
// GetState returns a snapshot of the current state for rendering
|
||||
func (dp *DetailedProgress) GetState() DetailedProgressState {
|
||||
dp.mu.RLock()
|
||||
defer dp.mu.RUnlock()
|
||||
|
||||
return DetailedProgressState{
|
||||
Description: dp.Description,
|
||||
Current: dp.Current,
|
||||
Total: dp.Total,
|
||||
Percent: dp.getPercentLocked(),
|
||||
Speed: dp.getSpeedLocked(),
|
||||
ETA: dp.getETALocked(),
|
||||
Elapsed: time.Since(dp.StartTime),
|
||||
Unit: dp.Unit,
|
||||
IsIndeterminate: dp.IsIndeterminate,
|
||||
IsComplete: dp.IsComplete,
|
||||
IsFailed: dp.IsFailed,
|
||||
ErrorMessage: dp.ErrorMessage,
|
||||
}
|
||||
}
|
||||
|
||||
func (dp *DetailedProgress) getPercentLocked() int {
|
||||
if dp.IsIndeterminate || dp.Total <= 0 {
|
||||
return 0
|
||||
}
|
||||
percent := int((dp.Current * 100) / dp.Total)
|
||||
if percent > 100 {
|
||||
return 100
|
||||
}
|
||||
return percent
|
||||
}
|
||||
|
||||
func (dp *DetailedProgress) getETALocked() time.Duration {
|
||||
if dp.IsIndeterminate || dp.Total <= 0 || dp.Current >= dp.Total {
|
||||
return 0
|
||||
}
|
||||
|
||||
speed := dp.getSpeedLocked()
|
||||
if speed <= 0 {
|
||||
return 0
|
||||
}
|
||||
|
||||
remaining := dp.Total - dp.Current
|
||||
seconds := float64(remaining) / speed
|
||||
return time.Duration(seconds) * time.Second
|
||||
}
|
||||
|
||||
// DetailedProgressState is an immutable snapshot for rendering
|
||||
type DetailedProgressState struct {
|
||||
Description string
|
||||
Current int64
|
||||
Total int64
|
||||
Percent int
|
||||
Speed float64 // bytes/sec
|
||||
ETA time.Duration
|
||||
Elapsed time.Duration
|
||||
Unit string
|
||||
IsIndeterminate bool
|
||||
IsComplete bool
|
||||
IsFailed bool
|
||||
ErrorMessage string
|
||||
}
|
||||
|
||||
// RenderProgressBar renders a TUI-friendly progress bar string
|
||||
// Returns something like: "Extracting archive [████████░░░░░░░░░░░░] 45% 12.5 MB/s ETA: 2m 30s"
|
||||
func (s DetailedProgressState) RenderProgressBar(width int) string {
|
||||
if s.IsIndeterminate {
|
||||
return s.renderIndeterminate()
|
||||
}
|
||||
|
||||
// Progress bar
|
||||
barWidth := 30
|
||||
if width < 80 {
|
||||
barWidth = 20
|
||||
}
|
||||
filled := (s.Percent * barWidth) / 100
|
||||
if filled > barWidth {
|
||||
filled = barWidth
|
||||
}
|
||||
|
||||
bar := strings.Repeat("█", filled) + strings.Repeat("░", barWidth-filled)
|
||||
|
||||
// Format bytes
|
||||
currentStr := FormatBytes(s.Current)
|
||||
totalStr := FormatBytes(s.Total)
|
||||
|
||||
// Format speed
|
||||
speedStr := ""
|
||||
if s.Speed > 0 {
|
||||
speedStr = fmt.Sprintf("%s/s", FormatBytes(int64(s.Speed)))
|
||||
}
|
||||
|
||||
// Format ETA
|
||||
etaStr := ""
|
||||
if s.ETA > 0 && !s.IsComplete {
|
||||
etaStr = fmt.Sprintf("ETA: %s", FormatDurationShort(s.ETA))
|
||||
}
|
||||
|
||||
// Build the line
|
||||
parts := []string{
|
||||
fmt.Sprintf("[%s]", bar),
|
||||
fmt.Sprintf("%3d%%", s.Percent),
|
||||
}
|
||||
|
||||
if s.Unit == "bytes" && s.Total > 0 {
|
||||
parts = append(parts, fmt.Sprintf("%s/%s", currentStr, totalStr))
|
||||
} else if s.Total > 0 {
|
||||
parts = append(parts, fmt.Sprintf("%d/%d", s.Current, s.Total))
|
||||
}
|
||||
|
||||
if speedStr != "" {
|
||||
parts = append(parts, speedStr)
|
||||
}
|
||||
if etaStr != "" {
|
||||
parts = append(parts, etaStr)
|
||||
}
|
||||
|
||||
return strings.Join(parts, " ")
|
||||
}
|
||||
|
||||
func (s DetailedProgressState) renderIndeterminate() string {
|
||||
elapsed := FormatDurationShort(s.Elapsed)
|
||||
return fmt.Sprintf("[spinner] %s Elapsed: %s", s.Description, elapsed)
|
||||
}
|
||||
|
||||
// RenderCompact renders a compact single-line progress string
|
||||
func (s DetailedProgressState) RenderCompact() string {
|
||||
if s.IsComplete {
|
||||
return fmt.Sprintf("[OK] %s completed in %s", s.Description, FormatDurationShort(s.Elapsed))
|
||||
}
|
||||
if s.IsFailed {
|
||||
return fmt.Sprintf("[FAIL] %s: %s", s.Description, s.ErrorMessage)
|
||||
}
|
||||
if s.IsIndeterminate {
|
||||
return fmt.Sprintf("[...] %s (%s)", s.Description, FormatDurationShort(s.Elapsed))
|
||||
}
|
||||
|
||||
return fmt.Sprintf("[%3d%%] %s - %s/%s", s.Percent, s.Description,
|
||||
FormatBytes(s.Current), FormatBytes(s.Total))
|
||||
}
|
||||
|
||||
// FormatBytes formats bytes in human-readable format
|
||||
func FormatBytes(b int64) string {
|
||||
const unit = 1024
|
||||
if b < unit {
|
||||
return fmt.Sprintf("%d B", b)
|
||||
}
|
||||
div, exp := int64(unit), 0
|
||||
for n := b / unit; n >= unit; n /= unit {
|
||||
div *= unit
|
||||
exp++
|
||||
}
|
||||
return fmt.Sprintf("%.1f %cB", float64(b)/float64(div), "KMGTPE"[exp])
|
||||
}
|
||||
|
||||
// FormatDurationShort formats duration in short form
|
||||
func FormatDurationShort(d time.Duration) string {
|
||||
if d < time.Second {
|
||||
return "<1s"
|
||||
}
|
||||
if d < time.Minute {
|
||||
return fmt.Sprintf("%ds", int(d.Seconds()))
|
||||
}
|
||||
if d < time.Hour {
|
||||
m := int(d.Minutes())
|
||||
s := int(d.Seconds()) % 60
|
||||
if s > 0 {
|
||||
return fmt.Sprintf("%dm %ds", m, s)
|
||||
}
|
||||
return fmt.Sprintf("%dm", m)
|
||||
}
|
||||
h := int(d.Hours())
|
||||
m := int(d.Minutes()) % 60
|
||||
return fmt.Sprintf("%dh %dm", h, m)
|
||||
}
|
||||
@@ -160,7 +160,7 @@ func (m DiagnoseViewModel) View() string {
|
||||
var s strings.Builder
|
||||
|
||||
// Header
|
||||
s.WriteString(titleStyle.Render("[SEARCH] Backup Diagnosis"))
|
||||
s.WriteString(titleStyle.Render("[CHECK] Backup Diagnosis"))
|
||||
s.WriteString("\n\n")
|
||||
|
||||
// Archive info
|
||||
@@ -204,132 +204,111 @@ func (m DiagnoseViewModel) View() string {
|
||||
func (m DiagnoseViewModel) renderSingleResult(result *restore.DiagnoseResult) string {
|
||||
var s strings.Builder
|
||||
|
||||
// Status Box
|
||||
s.WriteString("+--[ VALIDATION STATUS ]" + strings.Repeat("-", 37) + "+\n")
|
||||
// Validation Status
|
||||
s.WriteString(diagnoseHeaderStyle.Render("[STATUS] Validation"))
|
||||
s.WriteString("\n")
|
||||
|
||||
if result.IsValid {
|
||||
s.WriteString("| " + diagnosePassStyle.Render("[OK] VALID - Archive passed all checks") + strings.Repeat(" ", 18) + "|\n")
|
||||
s.WriteString(diagnosePassStyle.Render(" [OK] VALID - Archive passed all checks"))
|
||||
s.WriteString("\n")
|
||||
} else {
|
||||
s.WriteString("| " + diagnoseFailStyle.Render("[FAIL] INVALID - Archive has problems") + strings.Repeat(" ", 19) + "|\n")
|
||||
s.WriteString(diagnoseFailStyle.Render(" [FAIL] INVALID - Archive has problems"))
|
||||
s.WriteString("\n")
|
||||
}
|
||||
|
||||
if result.IsTruncated {
|
||||
s.WriteString("| " + diagnoseFailStyle.Render("[!] TRUNCATED - File is incomplete") + strings.Repeat(" ", 22) + "|\n")
|
||||
s.WriteString(diagnoseFailStyle.Render(" [!] TRUNCATED - File is incomplete"))
|
||||
s.WriteString("\n")
|
||||
}
|
||||
|
||||
if result.IsCorrupted {
|
||||
s.WriteString("| " + diagnoseFailStyle.Render("[!] CORRUPTED - File structure damaged") + strings.Repeat(" ", 18) + "|\n")
|
||||
s.WriteString(diagnoseFailStyle.Render(" [!] CORRUPTED - File structure damaged"))
|
||||
s.WriteString("\n")
|
||||
}
|
||||
|
||||
s.WriteString("+" + strings.Repeat("-", 60) + "+\n\n")
|
||||
s.WriteString("\n")
|
||||
|
||||
// Details Box
|
||||
// Details
|
||||
if result.Details != nil {
|
||||
s.WriteString("+--[ DETAILS ]" + strings.Repeat("-", 46) + "+\n")
|
||||
s.WriteString(diagnoseHeaderStyle.Render("[INFO] Details"))
|
||||
s.WriteString("\n")
|
||||
|
||||
if result.Details.HasPGDMPSignature {
|
||||
s.WriteString("| " + diagnosePassStyle.Render("[+]") + " PostgreSQL custom format (PGDMP)" + strings.Repeat(" ", 20) + "|\n")
|
||||
s.WriteString(diagnosePassStyle.Render(" [+]") + " PostgreSQL custom format (PGDMP)\n")
|
||||
}
|
||||
|
||||
if result.Details.HasSQLHeader {
|
||||
s.WriteString("| " + diagnosePassStyle.Render("[+]") + " PostgreSQL SQL header found" + strings.Repeat(" ", 25) + "|\n")
|
||||
s.WriteString(diagnosePassStyle.Render(" [+]") + " PostgreSQL SQL header found\n")
|
||||
}
|
||||
|
||||
if result.Details.GzipValid {
|
||||
s.WriteString("| " + diagnosePassStyle.Render("[+]") + " Gzip compression valid" + strings.Repeat(" ", 30) + "|\n")
|
||||
s.WriteString(diagnosePassStyle.Render(" [+]") + " Gzip compression valid\n")
|
||||
}
|
||||
|
||||
if result.Details.PgRestoreListable {
|
||||
tableInfo := fmt.Sprintf(" (%d tables)", result.Details.TableCount)
|
||||
padding := 36 - len(tableInfo)
|
||||
if padding < 0 {
|
||||
padding = 0
|
||||
}
|
||||
s.WriteString("| " + diagnosePassStyle.Render("[+]") + " pg_restore can list contents" + tableInfo + strings.Repeat(" ", padding) + "|\n")
|
||||
s.WriteString(diagnosePassStyle.Render(" [+]") + fmt.Sprintf(" pg_restore can list contents (%d tables)\n", result.Details.TableCount))
|
||||
}
|
||||
|
||||
if result.Details.CopyBlockCount > 0 {
|
||||
blockInfo := fmt.Sprintf("%d COPY blocks found", result.Details.CopyBlockCount)
|
||||
padding := 50 - len(blockInfo)
|
||||
if padding < 0 {
|
||||
padding = 0
|
||||
}
|
||||
s.WriteString("| [-] " + blockInfo + strings.Repeat(" ", padding) + "|\n")
|
||||
s.WriteString(fmt.Sprintf(" [-] %d COPY blocks found\n", result.Details.CopyBlockCount))
|
||||
}
|
||||
|
||||
if result.Details.UnterminatedCopy {
|
||||
s.WriteString("| " + diagnoseFailStyle.Render("[-]") + " Unterminated COPY: " + truncate(result.Details.LastCopyTable, 30) + strings.Repeat(" ", 5) + "|\n")
|
||||
s.WriteString(diagnoseFailStyle.Render(" [-]") + " Unterminated COPY: " + truncate(result.Details.LastCopyTable, 30) + "\n")
|
||||
}
|
||||
|
||||
if result.Details.ProperlyTerminated {
|
||||
s.WriteString("| " + diagnosePassStyle.Render("[+]") + " All COPY blocks properly terminated" + strings.Repeat(" ", 17) + "|\n")
|
||||
s.WriteString(diagnosePassStyle.Render(" [+]") + " All COPY blocks properly terminated\n")
|
||||
}
|
||||
|
||||
if result.Details.ExpandedSize > 0 {
|
||||
sizeInfo := fmt.Sprintf("Expanded: %s (%.1fx)", formatSize(result.Details.ExpandedSize), result.Details.CompressionRatio)
|
||||
padding := 50 - len(sizeInfo)
|
||||
if padding < 0 {
|
||||
padding = 0
|
||||
}
|
||||
s.WriteString("| [-] " + sizeInfo + strings.Repeat(" ", padding) + "|\n")
|
||||
s.WriteString(fmt.Sprintf(" [-] Expanded: %s (%.1fx)\n", formatSize(result.Details.ExpandedSize), result.Details.CompressionRatio))
|
||||
}
|
||||
|
||||
s.WriteString("+" + strings.Repeat("-", 60) + "+\n")
|
||||
s.WriteString("\n")
|
||||
}
|
||||
|
||||
// Errors Box
|
||||
// Errors
|
||||
if len(result.Errors) > 0 {
|
||||
s.WriteString("\n+--[ ERRORS ]" + strings.Repeat("-", 47) + "+\n")
|
||||
s.WriteString(diagnoseFailStyle.Render("[FAIL] Errors"))
|
||||
s.WriteString("\n")
|
||||
for i, e := range result.Errors {
|
||||
if i >= 5 {
|
||||
remaining := fmt.Sprintf("... and %d more errors", len(result.Errors)-5)
|
||||
padding := 56 - len(remaining)
|
||||
s.WriteString("| " + remaining + strings.Repeat(" ", padding) + "|\n")
|
||||
s.WriteString(fmt.Sprintf(" ... and %d more errors\n", len(result.Errors)-5))
|
||||
break
|
||||
}
|
||||
errText := truncate(e, 54)
|
||||
padding := 56 - len(errText)
|
||||
if padding < 0 {
|
||||
padding = 0
|
||||
s.WriteString(" " + truncate(e, 60) + "\n")
|
||||
}
|
||||
s.WriteString("| " + errText + strings.Repeat(" ", padding) + "|\n")
|
||||
}
|
||||
s.WriteString("+" + strings.Repeat("-", 60) + "+\n")
|
||||
s.WriteString("\n")
|
||||
}
|
||||
|
||||
// Warnings Box
|
||||
// Warnings
|
||||
if len(result.Warnings) > 0 {
|
||||
s.WriteString("\n+--[ WARNINGS ]" + strings.Repeat("-", 45) + "+\n")
|
||||
s.WriteString(diagnoseWarnStyle.Render("[WARN] Warnings"))
|
||||
s.WriteString("\n")
|
||||
for i, w := range result.Warnings {
|
||||
if i >= 3 {
|
||||
remaining := fmt.Sprintf("... and %d more warnings", len(result.Warnings)-3)
|
||||
padding := 56 - len(remaining)
|
||||
s.WriteString("| " + remaining + strings.Repeat(" ", padding) + "|\n")
|
||||
s.WriteString(fmt.Sprintf(" ... and %d more warnings\n", len(result.Warnings)-3))
|
||||
break
|
||||
}
|
||||
warnText := truncate(w, 54)
|
||||
padding := 56 - len(warnText)
|
||||
if padding < 0 {
|
||||
padding = 0
|
||||
s.WriteString(" " + truncate(w, 60) + "\n")
|
||||
}
|
||||
s.WriteString("| " + warnText + strings.Repeat(" ", padding) + "|\n")
|
||||
}
|
||||
s.WriteString("+" + strings.Repeat("-", 60) + "+\n")
|
||||
s.WriteString("\n")
|
||||
}
|
||||
|
||||
// Recommendations Box
|
||||
// Recommendations
|
||||
if !result.IsValid {
|
||||
s.WriteString("\n+--[ RECOMMENDATIONS ]" + strings.Repeat("-", 38) + "+\n")
|
||||
s.WriteString(diagnoseInfoStyle.Render("[HINT] Recommendations"))
|
||||
s.WriteString("\n")
|
||||
if result.IsTruncated {
|
||||
s.WriteString("| 1. Re-run backup with current version (v3.42.12+) |\n")
|
||||
s.WriteString("| 2. Check disk space on backup server |\n")
|
||||
s.WriteString("| 3. Verify network stability for remote backups |\n")
|
||||
s.WriteString(" 1. Re-run backup with current version (v3.42+)\n")
|
||||
s.WriteString(" 2. Check disk space on backup server\n")
|
||||
s.WriteString(" 3. Verify network stability for remote backups\n")
|
||||
}
|
||||
if result.IsCorrupted {
|
||||
s.WriteString("| 1. Verify backup was transferred completely |\n")
|
||||
s.WriteString("| 2. Try restoring from a previous backup |\n")
|
||||
s.WriteString(" 1. Verify backup was transferred completely\n")
|
||||
s.WriteString(" 2. Try restoring from a previous backup\n")
|
||||
}
|
||||
s.WriteString("+" + strings.Repeat("-", 60) + "+\n")
|
||||
}
|
||||
|
||||
return s.String()
|
||||
@@ -349,10 +328,8 @@ func (m DiagnoseViewModel) renderClusterResults() string {
|
||||
}
|
||||
}
|
||||
|
||||
s.WriteString(strings.Repeat("-", 60))
|
||||
s.WriteString("\n")
|
||||
s.WriteString(diagnoseHeaderStyle.Render(fmt.Sprintf("[STATS] CLUSTER SUMMARY: %d databases\n", len(m.results))))
|
||||
s.WriteString(strings.Repeat("-", 60))
|
||||
s.WriteString(diagnoseHeaderStyle.Render(fmt.Sprintf("[STATS] Cluster Summary: %d databases", len(m.results))))
|
||||
s.WriteString("\n\n")
|
||||
|
||||
if invalidCount == 0 {
|
||||
@@ -364,7 +341,7 @@ func (m DiagnoseViewModel) renderClusterResults() string {
|
||||
}
|
||||
|
||||
// List all dumps with status
|
||||
s.WriteString(diagnoseHeaderStyle.Render("Database Dumps:"))
|
||||
s.WriteString(diagnoseHeaderStyle.Render("[LIST] Database Dumps"))
|
||||
s.WriteString("\n")
|
||||
|
||||
// Show visible range based on cursor
|
||||
@@ -413,9 +390,7 @@ func (m DiagnoseViewModel) renderClusterResults() string {
|
||||
if m.cursor < len(m.results) {
|
||||
selected := m.results[m.cursor]
|
||||
s.WriteString("\n")
|
||||
s.WriteString(strings.Repeat("-", 60))
|
||||
s.WriteString("\n")
|
||||
s.WriteString(diagnoseHeaderStyle.Render("Selected: " + selected.FileName))
|
||||
s.WriteString(diagnoseHeaderStyle.Render("[INFO] Selected: " + selected.FileName))
|
||||
s.WriteString("\n\n")
|
||||
|
||||
// Show condensed details for selected
|
||||
|
||||
@@ -191,7 +191,7 @@ func (m HistoryViewModel) Update(msg tea.Msg) (tea.Model, tea.Cmd) {
|
||||
func (m HistoryViewModel) View() string {
|
||||
var s strings.Builder
|
||||
|
||||
header := titleStyle.Render("[HISTORY] Operation History")
|
||||
header := titleStyle.Render("[STATS] Operation History")
|
||||
s.WriteString(fmt.Sprintf("\n%s\n\n", header))
|
||||
|
||||
if len(m.history) == 0 {
|
||||
|
||||
@@ -285,7 +285,7 @@ func (m *MenuModel) View() string {
|
||||
var s string
|
||||
|
||||
// Header
|
||||
header := titleStyle.Render("[DB] Database Backup Tool - Interactive Menu")
|
||||
header := titleStyle.Render("Database Backup Tool - Interactive Menu")
|
||||
s += fmt.Sprintf("\n%s\n\n", header)
|
||||
|
||||
if len(m.dbTypes) > 0 {
|
||||
@@ -334,13 +334,13 @@ func (m *MenuModel) View() string {
|
||||
|
||||
// handleSingleBackup opens database selector for single backup
|
||||
func (m *MenuModel) handleSingleBackup() (tea.Model, tea.Cmd) {
|
||||
selector := NewDatabaseSelector(m.config, m.logger, m, m.ctx, "[DB] Single Database Backup", "single")
|
||||
selector := NewDatabaseSelector(m.config, m.logger, m, m.ctx, "[SELECT] Single Database Backup", "single")
|
||||
return selector, selector.Init()
|
||||
}
|
||||
|
||||
// handleSampleBackup opens database selector for sample backup
|
||||
func (m *MenuModel) handleSampleBackup() (tea.Model, tea.Cmd) {
|
||||
selector := NewDatabaseSelector(m.config, m.logger, m, m.ctx, "[STATS] Sample Database Backup", "sample")
|
||||
selector := NewDatabaseSelector(m.config, m.logger, m, m.ctx, "[SELECT] Sample Database Backup", "sample")
|
||||
return selector, selector.Init()
|
||||
}
|
||||
|
||||
@@ -356,7 +356,7 @@ func (m *MenuModel) handleClusterBackup() (tea.Model, tea.Cmd) {
|
||||
return executor, executor.Init()
|
||||
}
|
||||
confirm := NewConfirmationModelWithAction(m.config, m.logger, m,
|
||||
"[DB] Cluster Backup",
|
||||
"[CHECK] Cluster Backup",
|
||||
"This will backup ALL databases in the cluster. Continue?",
|
||||
func() (tea.Model, tea.Cmd) {
|
||||
executor := NewBackupExecution(m.config, m.logger, m, m.ctx, "cluster", "", 0)
|
||||
|
||||
@@ -6,6 +6,7 @@ import (
|
||||
"os/exec"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
tea "github.com/charmbracelet/bubbletea"
|
||||
@@ -45,6 +46,17 @@ type RestoreExecutionModel struct {
|
||||
spinnerFrame int
|
||||
spinnerFrames []string
|
||||
|
||||
// Detailed byte progress for schollz-style display
|
||||
bytesTotal int64
|
||||
bytesDone int64
|
||||
description string
|
||||
showBytes bool // True when we have real byte progress to show
|
||||
speed float64 // Rolling window speed in bytes/sec
|
||||
|
||||
// Database count progress (for cluster restore)
|
||||
dbTotal int
|
||||
dbDone int
|
||||
|
||||
// Results
|
||||
done bool
|
||||
cancelling bool // True when user has requested cancellation
|
||||
@@ -101,6 +113,9 @@ type restoreProgressMsg struct {
|
||||
phase string
|
||||
progress int
|
||||
detail string
|
||||
bytesTotal int64
|
||||
bytesDone int64
|
||||
description string
|
||||
}
|
||||
|
||||
type restoreCompleteMsg struct {
|
||||
@@ -109,6 +124,102 @@ type restoreCompleteMsg struct {
|
||||
elapsed time.Duration
|
||||
}
|
||||
|
||||
// sharedProgressState holds progress state that can be safely accessed from callbacks
|
||||
type sharedProgressState struct {
|
||||
mu sync.Mutex
|
||||
bytesTotal int64
|
||||
bytesDone int64
|
||||
description string
|
||||
hasUpdate bool
|
||||
|
||||
// Database count progress (for cluster restore)
|
||||
dbTotal int
|
||||
dbDone int
|
||||
|
||||
// Rolling window for speed calculation
|
||||
speedSamples []restoreSpeedSample
|
||||
}
|
||||
|
||||
type restoreSpeedSample struct {
|
||||
timestamp time.Time
|
||||
bytes int64
|
||||
}
|
||||
|
||||
// Package-level shared progress state for restore operations
|
||||
var (
|
||||
currentRestoreProgressMu sync.Mutex
|
||||
currentRestoreProgressState *sharedProgressState
|
||||
)
|
||||
|
||||
func setCurrentRestoreProgress(state *sharedProgressState) {
|
||||
currentRestoreProgressMu.Lock()
|
||||
defer currentRestoreProgressMu.Unlock()
|
||||
currentRestoreProgressState = state
|
||||
}
|
||||
|
||||
func clearCurrentRestoreProgress() {
|
||||
currentRestoreProgressMu.Lock()
|
||||
defer currentRestoreProgressMu.Unlock()
|
||||
currentRestoreProgressState = nil
|
||||
}
|
||||
|
||||
func getCurrentRestoreProgress() (bytesTotal, bytesDone int64, description string, hasUpdate bool, dbTotal, dbDone int, speed float64) {
|
||||
currentRestoreProgressMu.Lock()
|
||||
defer currentRestoreProgressMu.Unlock()
|
||||
|
||||
if currentRestoreProgressState == nil {
|
||||
return 0, 0, "", false, 0, 0, 0
|
||||
}
|
||||
|
||||
currentRestoreProgressState.mu.Lock()
|
||||
defer currentRestoreProgressState.mu.Unlock()
|
||||
|
||||
// Calculate rolling window speed
|
||||
speed = calculateRollingSpeed(currentRestoreProgressState.speedSamples)
|
||||
|
||||
return currentRestoreProgressState.bytesTotal, currentRestoreProgressState.bytesDone,
|
||||
currentRestoreProgressState.description, currentRestoreProgressState.hasUpdate,
|
||||
currentRestoreProgressState.dbTotal, currentRestoreProgressState.dbDone, speed
|
||||
}
|
||||
|
||||
// calculateRollingSpeed calculates speed from recent samples (last 5 seconds)
|
||||
func calculateRollingSpeed(samples []restoreSpeedSample) float64 {
|
||||
if len(samples) < 2 {
|
||||
return 0
|
||||
}
|
||||
|
||||
// Use samples from last 5 seconds for smoothed speed
|
||||
now := time.Now()
|
||||
cutoff := now.Add(-5 * time.Second)
|
||||
|
||||
var firstInWindow, lastInWindow *restoreSpeedSample
|
||||
for i := range samples {
|
||||
if samples[i].timestamp.After(cutoff) {
|
||||
if firstInWindow == nil {
|
||||
firstInWindow = &samples[i]
|
||||
}
|
||||
lastInWindow = &samples[i]
|
||||
}
|
||||
}
|
||||
|
||||
// Fall back to first and last if window is empty
|
||||
if firstInWindow == nil || lastInWindow == nil || firstInWindow == lastInWindow {
|
||||
firstInWindow = &samples[0]
|
||||
lastInWindow = &samples[len(samples)-1]
|
||||
}
|
||||
|
||||
elapsed := lastInWindow.timestamp.Sub(firstInWindow.timestamp).Seconds()
|
||||
if elapsed <= 0 {
|
||||
return 0
|
||||
}
|
||||
|
||||
bytesTransferred := lastInWindow.bytes - firstInWindow.bytes
|
||||
return float64(bytesTransferred) / elapsed
|
||||
}
|
||||
|
||||
// restoreProgressChannel allows sending progress updates from the restore goroutine
|
||||
type restoreProgressChannel chan restoreProgressMsg
|
||||
|
||||
func executeRestoreWithTUIProgress(parentCtx context.Context, cfg *config.Config, log logger.Logger, archive ArchiveInfo, targetDB string, cleanFirst, createIfMissing bool, restoreType string, cleanClusterFirst bool, existingDBs []string, saveDebugLog bool) tea.Cmd {
|
||||
return func() tea.Msg {
|
||||
// NO TIMEOUT for restore operations - a restore takes as long as it takes
|
||||
@@ -156,6 +267,48 @@ func executeRestoreWithTUIProgress(parentCtx context.Context, cfg *config.Config
|
||||
// STEP 2: Create restore engine with silent progress (no stdout interference with TUI)
|
||||
engine := restore.NewSilent(cfg, log, dbClient)
|
||||
|
||||
// Set up progress callback for detailed progress reporting
|
||||
// We use a shared pointer that can be queried by the TUI ticker
|
||||
progressState := &sharedProgressState{
|
||||
speedSamples: make([]restoreSpeedSample, 0, 100),
|
||||
}
|
||||
engine.SetProgressCallback(func(current, total int64, description string) {
|
||||
progressState.mu.Lock()
|
||||
defer progressState.mu.Unlock()
|
||||
progressState.bytesDone = current
|
||||
progressState.bytesTotal = total
|
||||
progressState.description = description
|
||||
progressState.hasUpdate = true
|
||||
|
||||
// Add speed sample for rolling window calculation
|
||||
progressState.speedSamples = append(progressState.speedSamples, restoreSpeedSample{
|
||||
timestamp: time.Now(),
|
||||
bytes: current,
|
||||
})
|
||||
// Keep only last 100 samples
|
||||
if len(progressState.speedSamples) > 100 {
|
||||
progressState.speedSamples = progressState.speedSamples[len(progressState.speedSamples)-100:]
|
||||
}
|
||||
})
|
||||
|
||||
// Set up database progress callback for cluster restore
|
||||
engine.SetDatabaseProgressCallback(func(done, total int, dbName string) {
|
||||
progressState.mu.Lock()
|
||||
defer progressState.mu.Unlock()
|
||||
progressState.dbDone = done
|
||||
progressState.dbTotal = total
|
||||
progressState.description = fmt.Sprintf("Restoring %s", dbName)
|
||||
progressState.hasUpdate = true
|
||||
// Clear byte progress when switching to db progress
|
||||
progressState.bytesTotal = 0
|
||||
progressState.bytesDone = 0
|
||||
})
|
||||
|
||||
// Store progress state in a package-level variable for the ticker to access
|
||||
// This is a workaround because tea messages can't be sent from callbacks
|
||||
setCurrentRestoreProgress(progressState)
|
||||
defer clearCurrentRestoreProgress()
|
||||
|
||||
// Enable debug logging if requested
|
||||
if saveDebugLog {
|
||||
// Generate debug log path using configured WorkDir
|
||||
@@ -165,9 +318,6 @@ func executeRestoreWithTUIProgress(parentCtx context.Context, cfg *config.Config
|
||||
log.Info("Debug logging enabled", "path", debugLogPath)
|
||||
}
|
||||
|
||||
// Set up progress callback (but it won't work in goroutine - progress is already sent via logs)
|
||||
// The TUI will just use spinner animation to show activity
|
||||
|
||||
// STEP 3: Execute restore based on type
|
||||
var restoreErr error
|
||||
if restoreType == "restore-cluster" {
|
||||
@@ -206,7 +356,29 @@ func (m RestoreExecutionModel) Update(msg tea.Msg) (tea.Model, tea.Cmd) {
|
||||
m.spinnerFrame = (m.spinnerFrame + 1) % len(m.spinnerFrames)
|
||||
m.elapsed = time.Since(m.startTime)
|
||||
|
||||
// Update status based on elapsed time to show progress
|
||||
// Poll shared progress state for real-time updates
|
||||
bytesTotal, bytesDone, description, hasUpdate, dbTotal, dbDone, speed := getCurrentRestoreProgress()
|
||||
if hasUpdate && bytesTotal > 0 {
|
||||
m.bytesTotal = bytesTotal
|
||||
m.bytesDone = bytesDone
|
||||
m.description = description
|
||||
m.showBytes = true
|
||||
m.speed = speed
|
||||
|
||||
// Update status to reflect actual progress
|
||||
m.status = description
|
||||
m.phase = "Extracting"
|
||||
m.progress = int((bytesDone * 100) / bytesTotal)
|
||||
} else if hasUpdate && dbTotal > 0 {
|
||||
// Database count progress for cluster restore
|
||||
m.dbTotal = dbTotal
|
||||
m.dbDone = dbDone
|
||||
m.showBytes = false
|
||||
m.status = fmt.Sprintf("Restoring database %d of %d...", dbDone+1, dbTotal)
|
||||
m.phase = "Restore"
|
||||
m.progress = int((dbDone * 100) / dbTotal)
|
||||
} else {
|
||||
// Fallback: Update status based on elapsed time to show progress
|
||||
// This provides visual feedback even though we don't have real-time progress
|
||||
elapsedSec := int(m.elapsed.Seconds())
|
||||
|
||||
@@ -241,6 +413,7 @@ func (m RestoreExecutionModel) Update(msg tea.Msg) (tea.Model, tea.Cmd) {
|
||||
m.phase = "Restore"
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return m, restoreTickCmd()
|
||||
}
|
||||
@@ -250,6 +423,15 @@ func (m RestoreExecutionModel) Update(msg tea.Msg) (tea.Model, tea.Cmd) {
|
||||
m.status = msg.status
|
||||
m.phase = msg.phase
|
||||
m.progress = msg.progress
|
||||
|
||||
// Update byte-level progress if available
|
||||
if msg.bytesTotal > 0 {
|
||||
m.bytesTotal = msg.bytesTotal
|
||||
m.bytesDone = msg.bytesDone
|
||||
m.description = msg.description
|
||||
m.showBytes = true
|
||||
}
|
||||
|
||||
if msg.detail != "" {
|
||||
m.details = append(m.details, msg.detail)
|
||||
// Keep only last 5 details
|
||||
@@ -321,9 +503,9 @@ func (m RestoreExecutionModel) View() string {
|
||||
s.Grow(512) // Pre-allocate estimated capacity for better performance
|
||||
|
||||
// Title
|
||||
title := "[RESTORE] Restoring Database"
|
||||
title := "[EXEC] Restoring Database"
|
||||
if m.restoreType == "restore-cluster" {
|
||||
title = "[RESTORE] Restoring Cluster"
|
||||
title = "[EXEC] Restoring Cluster"
|
||||
}
|
||||
s.WriteString(titleStyle.Render(title))
|
||||
s.WriteString("\n\n")
|
||||
@@ -356,19 +538,39 @@ func (m RestoreExecutionModel) View() string {
|
||||
// Show progress
|
||||
s.WriteString(fmt.Sprintf("Phase: %s\n", m.phase))
|
||||
|
||||
// Show status with rotating spinner (unified indicator for all operations)
|
||||
// Show detailed progress bar when we have byte-level information
|
||||
// In this case, hide the spinner for cleaner display
|
||||
if m.showBytes && m.bytesTotal > 0 {
|
||||
// Status line without spinner (progress bar provides activity indication)
|
||||
s.WriteString(fmt.Sprintf("Status: %s\n", m.status))
|
||||
s.WriteString("\n")
|
||||
|
||||
// Render schollz-style progress bar with bytes, rolling speed, ETA
|
||||
s.WriteString(renderDetailedProgressBarWithSpeed(m.bytesDone, m.bytesTotal, m.speed))
|
||||
s.WriteString("\n\n")
|
||||
} else if m.dbTotal > 0 {
|
||||
// Database count progress for cluster restore
|
||||
spinner := m.spinnerFrames[m.spinnerFrame]
|
||||
s.WriteString(fmt.Sprintf("Status: %s %s\n", spinner, m.status))
|
||||
s.WriteString("\n")
|
||||
|
||||
// Show database progress bar
|
||||
s.WriteString(renderDatabaseProgressBar(m.dbDone, m.dbTotal))
|
||||
s.WriteString("\n\n")
|
||||
} else {
|
||||
// Show status with rotating spinner (for phases without detailed progress)
|
||||
spinner := m.spinnerFrames[m.spinnerFrame]
|
||||
s.WriteString(fmt.Sprintf("Status: %s %s\n", spinner, m.status))
|
||||
s.WriteString("\n")
|
||||
|
||||
// Only show progress bar for single database restore
|
||||
// Cluster restore uses spinner only (consistent with CLI behavior)
|
||||
if m.restoreType == "restore-single" {
|
||||
// Fallback to simple progress bar for single database restore
|
||||
progressBar := renderProgressBar(m.progress)
|
||||
s.WriteString(progressBar)
|
||||
s.WriteString(fmt.Sprintf(" %d%%\n", m.progress))
|
||||
s.WriteString("\n")
|
||||
}
|
||||
}
|
||||
|
||||
// Elapsed time
|
||||
s.WriteString(fmt.Sprintf("Elapsed: %s\n", formatDuration(m.elapsed)))
|
||||
@@ -390,6 +592,92 @@ func renderProgressBar(percent int) string {
|
||||
return successStyle.Render(bar) + infoStyle.Render(empty)
|
||||
}
|
||||
|
||||
// renderDetailedProgressBar renders a schollz-style progress bar with bytes, speed, and ETA
|
||||
// Uses elapsed time for speed calculation (fallback)
|
||||
func renderDetailedProgressBar(done, total int64, elapsed time.Duration) string {
|
||||
speed := 0.0
|
||||
if elapsed.Seconds() > 0 {
|
||||
speed = float64(done) / elapsed.Seconds()
|
||||
}
|
||||
return renderDetailedProgressBarWithSpeed(done, total, speed)
|
||||
}
|
||||
|
||||
// renderDetailedProgressBarWithSpeed renders a schollz-style progress bar with pre-calculated rolling speed
|
||||
func renderDetailedProgressBarWithSpeed(done, total int64, speed float64) string {
|
||||
var s strings.Builder
|
||||
|
||||
// Calculate percentage
|
||||
percent := 0
|
||||
if total > 0 {
|
||||
percent = int((done * 100) / total)
|
||||
if percent > 100 {
|
||||
percent = 100
|
||||
}
|
||||
}
|
||||
|
||||
// Render progress bar
|
||||
width := 30
|
||||
filled := (percent * width) / 100
|
||||
barFilled := strings.Repeat("█", filled)
|
||||
barEmpty := strings.Repeat("░", width-filled)
|
||||
|
||||
s.WriteString(successStyle.Render("["))
|
||||
s.WriteString(successStyle.Render(barFilled))
|
||||
s.WriteString(infoStyle.Render(barEmpty))
|
||||
s.WriteString(successStyle.Render("]"))
|
||||
|
||||
// Percentage
|
||||
s.WriteString(fmt.Sprintf(" %3d%%", percent))
|
||||
|
||||
// Bytes progress
|
||||
s.WriteString(fmt.Sprintf(" %s / %s", FormatBytes(done), FormatBytes(total)))
|
||||
|
||||
// Speed display (using rolling window speed)
|
||||
if speed > 0 {
|
||||
s.WriteString(fmt.Sprintf(" %s/s", FormatBytes(int64(speed))))
|
||||
|
||||
// ETA calculation based on rolling speed
|
||||
if done < total {
|
||||
remaining := total - done
|
||||
etaSeconds := float64(remaining) / speed
|
||||
eta := time.Duration(etaSeconds) * time.Second
|
||||
s.WriteString(fmt.Sprintf(" ETA: %s", FormatDurationShort(eta)))
|
||||
}
|
||||
}
|
||||
|
||||
return s.String()
|
||||
}
|
||||
|
||||
// renderDatabaseProgressBar renders a progress bar for database count (cluster restore)
|
||||
func renderDatabaseProgressBar(done, total int) string {
|
||||
var s strings.Builder
|
||||
|
||||
// Calculate percentage
|
||||
percent := 0
|
||||
if total > 0 {
|
||||
percent = (done * 100) / total
|
||||
if percent > 100 {
|
||||
percent = 100
|
||||
}
|
||||
}
|
||||
|
||||
// Render progress bar
|
||||
width := 30
|
||||
filled := (percent * width) / 100
|
||||
barFilled := strings.Repeat("█", filled)
|
||||
barEmpty := strings.Repeat("░", width-filled)
|
||||
|
||||
s.WriteString(successStyle.Render("["))
|
||||
s.WriteString(successStyle.Render(barFilled))
|
||||
s.WriteString(infoStyle.Render(barEmpty))
|
||||
s.WriteString(successStyle.Render("]"))
|
||||
|
||||
// Count and percentage
|
||||
s.WriteString(fmt.Sprintf(" %3d%% %d / %d databases", percent, done, total))
|
||||
|
||||
return s.String()
|
||||
}
|
||||
|
||||
// formatDuration formats duration in human readable format
|
||||
func formatDuration(d time.Duration) string {
|
||||
if d < time.Minute {
|
||||
|
||||
@@ -339,9 +339,9 @@ func (m RestorePreviewModel) View() string {
|
||||
var s strings.Builder
|
||||
|
||||
// Title
|
||||
title := "Restore Preview"
|
||||
title := "[CHECK] Restore Preview"
|
||||
if m.mode == "restore-cluster" {
|
||||
title = "Cluster Restore Preview"
|
||||
title = "[CHECK] Cluster Restore Preview"
|
||||
}
|
||||
s.WriteString(titleStyle.Render(title))
|
||||
s.WriteString("\n\n")
|
||||
|
||||
@@ -688,7 +688,7 @@ func (m SettingsModel) View() string {
|
||||
var b strings.Builder
|
||||
|
||||
// Header
|
||||
header := titleStyle.Render("[CFG] Configuration Settings")
|
||||
header := titleStyle.Render("[CONFIG] Configuration Settings")
|
||||
b.WriteString(fmt.Sprintf("\n%s\n\n", header))
|
||||
|
||||
// Settings list
|
||||
@@ -747,7 +747,7 @@ func (m SettingsModel) View() string {
|
||||
// Current configuration summary
|
||||
if !m.editing {
|
||||
b.WriteString("\n")
|
||||
b.WriteString(infoStyle.Render("[LOG] Current Configuration:"))
|
||||
b.WriteString(infoStyle.Render("[INFO] Current Configuration"))
|
||||
b.WriteString("\n")
|
||||
|
||||
summary := []string{
|
||||
|
||||
@@ -173,7 +173,7 @@ func (m StatusViewModel) View() string {
|
||||
s.WriteString(errorStyle.Render(fmt.Sprintf("[FAIL] Error: %v\n", m.err)))
|
||||
s.WriteString("\n")
|
||||
} else {
|
||||
s.WriteString("Connection Status:\n")
|
||||
s.WriteString("[CONN] Connection Status\n")
|
||||
if m.connected {
|
||||
s.WriteString(successStyle.Render(" [+] Connected\n"))
|
||||
} else {
|
||||
@@ -181,11 +181,12 @@ func (m StatusViewModel) View() string {
|
||||
}
|
||||
s.WriteString("\n")
|
||||
|
||||
s.WriteString(fmt.Sprintf("Database Type: %s (%s)\n", m.config.DisplayDatabaseType(), m.config.DatabaseType))
|
||||
s.WriteString(fmt.Sprintf("Host: %s:%d\n", m.config.Host, m.config.Port))
|
||||
s.WriteString(fmt.Sprintf("User: %s\n", m.config.User))
|
||||
s.WriteString(fmt.Sprintf("Backup Directory: %s\n", m.config.BackupDir))
|
||||
s.WriteString(fmt.Sprintf("Version: %s\n\n", m.dbVersion))
|
||||
s.WriteString("[INFO] Server Details\n")
|
||||
s.WriteString(fmt.Sprintf(" Database Type: %s (%s)\n", m.config.DisplayDatabaseType(), m.config.DatabaseType))
|
||||
s.WriteString(fmt.Sprintf(" Host: %s:%d\n", m.config.Host, m.config.Port))
|
||||
s.WriteString(fmt.Sprintf(" User: %s\n", m.config.User))
|
||||
s.WriteString(fmt.Sprintf(" Backup Directory: %s\n", m.config.BackupDir))
|
||||
s.WriteString(fmt.Sprintf(" Version: %s\n\n", m.dbVersion))
|
||||
|
||||
if m.dbCount > 0 {
|
||||
s.WriteString(fmt.Sprintf("Databases Found: %s\n", successStyle.Render(fmt.Sprintf("%d", m.dbCount))))
|
||||
|
||||
@@ -120,12 +120,36 @@ var ShortcutStyle = lipgloss.NewStyle().
|
||||
// =============================================================================
|
||||
// HELPER PREFIXES (no emoticons)
|
||||
// =============================================================================
|
||||
// Convention for TUI titles/headers:
|
||||
// [CHECK] - Verification/diagnosis screens
|
||||
// [STATS] - Statistics/status screens
|
||||
// [SELECT] - Selection/browser screens
|
||||
// [EXEC] - Execution/running screens
|
||||
// [CONFIG] - Configuration/settings screens
|
||||
//
|
||||
// Convention for status messages:
|
||||
// [OK] - Success
|
||||
// [FAIL] - Error/failure
|
||||
// [WAIT] - In progress
|
||||
// [WARN] - Warning
|
||||
// [INFO] - Information
|
||||
|
||||
const (
|
||||
// Title prefixes (for view headers)
|
||||
PrefixCheck = "[CHECK]"
|
||||
PrefixStats = "[STATS]"
|
||||
PrefixSelect = "[SELECT]"
|
||||
PrefixExec = "[EXEC]"
|
||||
PrefixConfig = "[CONFIG]"
|
||||
|
||||
// Status prefixes
|
||||
PrefixOK = "[OK]"
|
||||
PrefixFail = "[FAIL]"
|
||||
PrefixWarn = "[!]"
|
||||
PrefixInfo = "[i]"
|
||||
PrefixWait = "[WAIT]"
|
||||
PrefixWarn = "[WARN]"
|
||||
PrefixInfo = "[INFO]"
|
||||
|
||||
// List item prefixes
|
||||
PrefixPlus = "[+]"
|
||||
PrefixMinus = "[-]"
|
||||
PrefixArrow = ">"
|
||||
|
||||
Reference in New Issue
Block a user