Backup and Restore Guide

package main

import (
    "context"
    "database/sql"
    "fmt"
    "log"
    "time"

    _ "geodedb.com/geode"
)

func createOnlineBackup(db *sql.DB, backupPath string) error {
    ctx := context.Background()

    // Start backup
    _, err := db.ExecContext(ctx,
        "CALL geode.backup.create($path, $compress)",
        backupPath, true)
    if err != nil {
        return fmt.Errorf("backup failed: %w", err)
    }

    // Verify backup
    rows, err := db.QueryContext(ctx,
        "CALL geode.backup.verify($path)",
        backupPath)
    if err != nil {
        return fmt.Errorf("verification failed: %w", err)
    }
    defer rows.Close()

    var valid bool
    var checksum string
    if rows.Next() {
        rows.Scan(&valid, &checksum)
        if !valid {
            return fmt.Errorf("backup verification failed")
        }
        log.Printf("Backup verified: checksum=%s", checksum)
    }

    return nil
}

func main() {
    db, err := sql.Open("geode", "localhost:3141")
    if err != nil {
        log.Fatal(err)
    }
    defer db.Close()

    backupPath := fmt.Sprintf("/backups/geode-%s.backup",
        time.Now().Format("20060102-150405"))

    if err := createOnlineBackup(db, backupPath); err != nil {
        log.Fatalf("Backup failed: %v", err)
    }

    log.Printf("Backup created successfully: %s", backupPath)
}

import asyncio
import os
import json
from datetime import datetime
from geode_client import Client

class IncrementalBackupManager:
    def __init__(self, backup_dir: str):
        self.backup_dir = backup_dir
        self.full_dir = os.path.join(backup_dir, "full")
        self.incr_dir = os.path.join(backup_dir, "incremental")
        self.state_file = os.path.join(backup_dir, "backup_state.json")

        os.makedirs(self.full_dir, exist_ok=True)
        os.makedirs(self.incr_dir, exist_ok=True)

    def _load_state(self) -> dict:
        try:
            with open(self.state_file, 'r') as f:
                return json.load(f)
        except FileNotFoundError:
            return {"last_full": None, "last_lsn": 0}

    def _save_state(self, state: dict):
        with open(self.state_file, 'w') as f:
            json.dump(state, f, indent=2)

    async def create_full_backup(self) -> str:
        """Create a new full backup."""
        client = Client(host="localhost", port=3141, skip_verify=True)

        timestamp = datetime.now().strftime("%Y%m%d-%H%M%S")
        backup_path = os.path.join(self.full_dir, f"full-{timestamp}.backup")

        async with client.connection() as conn:
            # Create full backup
            result, _ = await conn.query(
                """CALL geode.backup.full($path, {
                    include_metadata: true,
                    compression: 'gzip'
                })""",
                {"path": backup_path}
            )

            if result.rows:
                lsn = result.rows[0]['end_lsn'].as_int

                # Update state
                state = self._load_state()
                state["last_full"] = backup_path
                state["last_lsn"] = lsn
                self._save_state(state)

                print(f"Full backup created: {backup_path}")
                print(f"  End LSN: {lsn}")

                return backup_path

        return None

    async def create_incremental_backup(self) -> str:
        """Create an incremental backup since last backup."""
        state = self._load_state()

        if not state["last_full"]:
            print("No full backup exists. Creating full backup first...")
            return await self.create_full_backup()

        client = Client(host="localhost", port=3141, skip_verify=True)

        timestamp = datetime.now().strftime("%Y%m%d-%H%M%S")
        backup_path = os.path.join(self.incr_dir, f"incr-{timestamp}.backup")

        async with client.connection() as conn:
            # Create incremental backup
            result, _ = await conn.query(
                """CALL geode.backup.incremental($path, $from_lsn, {
                    compression: 'gzip'
                })""",
                {"path": backup_path, "from_lsn": state["last_lsn"]}
            )

            if result.rows:
                row = result.rows[0]
                changes = row['change_count'].as_int
                end_lsn = row['end_lsn'].as_int

                if changes == 0:
                    print("No changes since last backup")
                    os.remove(backup_path)
                    return None

                # Update state
                state["last_lsn"] = end_lsn
                self._save_state(state)

                print(f"Incremental backup created: {backup_path}")
                print(f"  Changes: {changes:,}")
                print(f"  End LSN: {end_lsn}")

                return backup_path

        return None

async def main():
    manager = IncrementalBackupManager("/backups/geode")

    # Weekly: Create full backup
    # await manager.create_full_backup()

    # Daily: Create incremental backup
    await manager.create_incremental_backup()

asyncio.run(main())

import asyncio
from datetime import datetime
from geode_client import Client

async def point_in_time_recovery(
    backup_path: str,
    wal_dir: str,
    target_time: datetime,
    output_dir: str
):
    """Restore database to a specific point in time."""
    client = Client(host="localhost", port=3141, skip_verify=True)

    async with client.connection() as conn:
        print(f"Starting point-in-time recovery...")
        print(f"  Base backup: {backup_path}")
        print(f"  Target time: {target_time.isoformat()}")

        # Initiate PITR
        result, _ = await conn.query(
            """CALL geode.restore.point_in_time($backup, $wal_dir, $target, $output, {
                verify: true,
                create_recovery_conf: true
            })""",
            {
                "backup": backup_path,
                "wal_dir": wal_dir,
                "target": target_time.isoformat(),
                "output": output_dir
            }
        )

        if result.rows:
            info = result.rows[0]
            print(f"Recovery complete:")
            print(f"  Restored to: {info['restored_to'].as_string}")
            print(f"  WAL files applied: {info['wal_files_applied'].as_int}")
            print(f"  Transactions recovered: {info['transactions_recovered'].as_int}")

            return True

    return False

async def main():
    # Restore to 2:30 PM today
    target = datetime(2026, 1, 28, 14, 30, 0)

    await point_in_time_recovery(
        backup_path="/backups/full/geode-full-20260125.backup",
        wal_dir="/backups/wal",
        target_time=target,
        output_dir="/var/lib/geode/data-recovered"
    )

asyncio.run(main())

import asyncio
from geode_client import Client

async def verify_backup(backup_path: str) -> bool:
    """Comprehensively verify a backup file."""
    client = Client(host="localhost", port=3141, skip_verify=True)

    async with client.connection() as conn:
        print(f"Verifying backup: {backup_path}")

        # Basic integrity check
        result, _ = await conn.query(
            """CALL geode.backup.verify($path, {
                check_checksums: true,
                check_consistency: true,
                check_indexes: true
            })""",
            {"path": backup_path}
        )

        if not result.rows:
            print("Verification failed: No result")
            return False

        info = result.rows[0]

        checks = {
            "File integrity": info['file_valid'].as_bool,
            "Checksum match": info['checksum_valid'].as_bool,
            "Data consistency": info['data_consistent'].as_bool,
            "Index integrity": info['indexes_valid'].as_bool,
        }

        all_passed = True
        for check_name, passed in checks.items():
            status = "PASS" if passed else "FAIL"
            print(f"  {check_name}: {status}")
            if not passed:
                all_passed = False

        if all_passed:
            print(f"\nBackup verification: PASSED")
            print(f"  Checksum: {info['checksum'].as_string}")
            print(f"  Node count: {info['node_count'].as_int:,}")
            print(f"  Relationship count: {info['relationship_count'].as_int:,}")
        else:
            print(f"\nBackup verification: FAILED")

        return all_passed

async def verify_all_backups(backup_dir: str):
    """Verify all backups in a directory."""
    import os

    results = []
    for filename in os.listdir(backup_dir):
        if filename.endswith('.backup'):
            path = os.path.join(backup_dir, filename)
            valid = await verify_backup(path)
            results.append((filename, valid))
            print()

    print("=" * 50)
    print("SUMMARY")
    print("=" * 50)
    for filename, valid in results:
        status = "OK" if valid else "FAILED"
        print(f"  {filename}: {status}")

asyncio.run(verify_all_backups("/backups/full"))

import asyncio
import os
from geode_client import Client

async def restore_database(
    backup_path: str,
    output_dir: str,
    incremental_backups: list = None
):
    """Restore database from backup files."""
    client = Client(host="localhost", port=3141, skip_verify=True)

    # Ensure output directory exists
    os.makedirs(output_dir, exist_ok=True)

    async with client.connection() as conn:
        # Restore full backup
        print(f"Restoring full backup: {backup_path}")

        result, _ = await conn.query(
            """CALL geode.restore.full($backup, $output, {
                verify: true,
                overwrite: true
            })""",
            {"backup": backup_path, "output": output_dir}
        )

        if result.rows:
            info = result.rows[0]
            print(f"  Nodes restored: {info['node_count'].as_int:,}")
            print(f"  Relationships restored: {info['relationship_count'].as_int:,}")

        # Apply incremental backups
        if incremental_backups:
            for incr_backup in incremental_backups:
                print(f"Applying incremental: {incr_backup}")

                result, _ = await conn.query(
                    """CALL geode.restore.incremental($backup, $output, {
                        verify: true
                    })""",
                    {"backup": incr_backup, "output": output_dir}
                )

                if result.rows:
                    changes = result.rows[0]['changes_applied'].as_int
                    print(f"  Changes applied: {changes:,}")

        print(f"\nRestore complete. Data directory: {output_dir}")

async def main():
    await restore_database(
        backup_path="/backups/full/geode-full-20260125.backup",
        output_dir="/var/lib/geode/data-restored",
        incremental_backups=[
            "/backups/incremental/geode-incr-20260126.backup",
            "/backups/incremental/geode-incr-20260127.backup",
            "/backups/incremental/geode-incr-20260128.backup",
        ]
    )

asyncio.run(main())

import asyncio
import boto3
from datetime import datetime
from geode_client import Client

class MultiRegionBackupManager:
    def __init__(self):
        self.primary_region = "us-east-1"
        self.backup_regions = ["us-west-2", "eu-west-1"]
        self.s3_clients = {
            region: boto3.client('s3', region_name=region)
            for region in [self.primary_region] + self.backup_regions
        }

    async def create_and_replicate_backup(self):
        """Create backup and replicate to multiple regions."""
        client = Client(host="localhost", port=3141, skip_verify=True)

        timestamp = datetime.now().strftime("%Y%m%d-%H%M%S")
        local_path = f"/tmp/geode-backup-{timestamp}.backup"
        s3_key = f"backups/geode/full-{timestamp}.backup"

        async with client.connection() as conn:
            # Create local backup
            print("Creating backup...")
            await conn.execute(
                "CALL geode.backup.full($path, {compression: 'gzip'})",
                {"path": local_path}
            )

            # Upload to primary region
            print(f"Uploading to {self.primary_region}...")
            self.s3_clients[self.primary_region].upload_file(
                local_path,
                f"geode-backups-{self.primary_region}",
                s3_key
            )

            # Replicate to backup regions
            for region in self.backup_regions:
                print(f"Replicating to {region}...")

                # Copy from primary to backup region
                self.s3_clients[region].copy_object(
                    CopySource={
                        'Bucket': f"geode-backups-{self.primary_region}",
                        'Key': s3_key
                    },
                    Bucket=f"geode-backups-{region}",
                    Key=s3_key
                )

            print(f"Backup replicated to {len(self.backup_regions)} regions")

            # Clean up local file
            import os
            os.remove(local_path)

    async def restore_from_nearest_region(self, target_dir: str):
        """Restore from the nearest available region."""
        # Check backup availability in each region
        for region in [self.primary_region] + self.backup_regions:
            bucket = f"geode-backups-{region}"
            try:
                response = self.s3_clients[region].list_objects_v2(
                    Bucket=bucket,
                    Prefix="backups/geode/full-",
                    MaxKeys=1
                )

                if response.get('Contents'):
                    latest = sorted(
                        response['Contents'],
                        key=lambda x: x['LastModified'],
                        reverse=True
                    )[0]

                    print(f"Restoring from {region}: {latest['Key']}")

                    # Download backup
                    local_path = f"/tmp/restore-{region}.backup"
                    self.s3_clients[region].download_file(
                        bucket,
                        latest['Key'],
                        local_path
                    )

                    # Restore
                    client = Client(host="localhost", port=3141, skip_verify=True)
                    async with client.connection() as conn:
                        await conn.execute(
                            "CALL geode.restore.full($path, $output)",
                            {"path": local_path, "output": target_dir}
                        )

                    print(f"Restore complete from {region}")
                    return True

            except Exception as e:
                print(f"Region {region} unavailable: {e}")
                continue

        print("No backup available in any region!")
        return False

async def main():
    manager = MultiRegionBackupManager()

    # Create and replicate backup
    await manager.create_and_replicate_backup()

asyncio.run(main())