migration

func (*SQLExecutor) ExecContext ¶

func (e *SQLExecutor) ExecContext(ctx context.Context, query string, args ...interface{}) (Result, error)

func (*SQLExecutor) QueryContext ¶

func (e *SQLExecutor) QueryContext(ctx context.Context, query string, args ...interface{}) (Rows, error)

func (*SQLExecutor) QueryRowContext ¶

func (e *SQLExecutor) QueryRowContext(ctx context.Context, query string, args ...interface{}) Row

func (*Tracker) ClaimPartition ¶

func (t *Tracker) ClaimPartition(ctx context.Context, podID string, claimTimeout time.Duration) (*Partition, error)

ClaimPartition claims the next pending partition for processing.

This method uses SELECT FOR UPDATE SKIP LOCKED to safely claim a partition in a concurrent environment. If multiple workers call this simultaneously, each will get a different partition (or ErrNoPendingPartition if none available).

The claim is conditional: it only succeeds if the partition is still pending. This prevents race conditions where one worker reads a pending partition but another worker claims it first.

Parameters:

• ctx: Context for the database operation
• podID: Identifier for the pod claiming this partition (e.g., "migration-worker-abc123")
• claimTimeout: Duration after which a claim is considered stale (for crash recovery)

Returns:

• *Partition: The claimed partition with status set to 'in_progress'
• error: ErrNoPendingPartition if no partitions are available, or a database error

Example usage:

partition, err := tracker.ClaimPartition(ctx, "migration-worker-pod-123", 30*time.Minute)
if err != nil {
    if errors.Is(err, migration.ErrNoPendingPartition) {
        log.Println("No work available - sleeping...")
        return nil
    }
    return fmt.Errorf("failed to claim partition: %w", err)
}
log.Printf("Claimed partition: %s\n", partition.PartitionID)
// Process the partition...

func (*Tracker) CreatePartition ¶

func (t *Tracker) CreatePartition(ctx context.Context, partitionID string) error

CreatePartition creates a new partition in pending status.

This is used during the seed phase to populate the migration_progress table with all partitions that need to be processed.

Parameters:

• ctx: Context for the database operation
• partitionID: The partition identifier

Returns:

• error: Database error, or error if partition already exists

Example usage:

err := tracker.CreatePartition(ctx, "day=2023-01-01")
if err != nil {
    return fmt.Errorf("failed to create partition: %w", err)
}

func (*Tracker) GetMigrationStatus ¶

func (t *Tracker) GetMigrationStatus(ctx context.Context) (map[PartitionStatus]int64, error)

GetMigrationStatus returns overall migration progress statistics.

Returns counts for each status, useful for monitoring and dashboards.

Parameters:

• ctx: Context for the database operation

Returns:

• map[PartitionStatus]int64: Count of partitions in each status
• error: Database error

Example usage:

stats, err := tracker.GetMigrationStatus(ctx)
if err != nil {
    return fmt.Errorf("failed to get migration status: %w", err)
}
log.Printf("Pending: %d, In Progress: %d, Completed: %d, Failed: %d\n",
    stats[migration.StatusPending],
    stats[migration.StatusInProgress],
    stats[migration.StatusCompleted],
    stats[migration.StatusFailed])

func (*Tracker) GetPartition ¶

func (t *Tracker) GetPartition(ctx context.Context, partitionID string) (*Partition, error)

GetPartition retrieves a single partition by ID.

Parameters:

• ctx: Context for the database operation
• partitionID: The partition identifier

Returns:

• *Partition: The partition, or nil if not found
• error: Database error (not ErrPartitionNotFound - use nil check instead)

Example usage:

partition, err := tracker.GetPartition(ctx, "day=2023-01-01")
if err != nil {
    return fmt.Errorf("failed to get partition: %w", err)
}
if partition == nil {
    log.Println("Partition not found")
    return nil
}

func (*Tracker) GetStaleClaims ¶

func (t *Tracker) GetStaleClaims(ctx context.Context, timeout time.Duration) ([]*Partition, error)

GetStaleClaims retrieves partitions that have been in_progress for longer than the timeout.

This method is used on worker startup to detect and recover from crashes. Any partition that has been claimed for longer than the timeout is considered abandoned and should be released back to pending.

Parameters:

• ctx: Context for the database operation
• timeout: Duration after which a claim is considered stale

Returns:

• []*Partition: List of stale partitions
• error: Database error

Example usage:

staleClaims, err := tracker.GetStaleClaims(ctx, 30*time.Minute)
if err != nil {
    return fmt.Errorf("failed to get stale claims: %w", err)
}
for _, claim := range staleClaims {
    log.Printf("Releasing stale claim: %s (claimed by %s since %s)\n",
        claim.PartitionID, *claim.ClaimedBy, claim.ClaimedAt)
    err := tracker.ReleasePartition(ctx, claim.PartitionID, *claim.ClaimedBy)
    if err != nil {
        log.Printf("Failed to release: %v\n", err)
    }
}

func (*Tracker) ListPartitions ¶

func (t *Tracker) ListPartitions(ctx context.Context, status PartitionStatus) ([]*Partition, error)

ListPartitions retrieves all partitions with optional filtering.

Parameters:

• ctx: Context for the database operation
• status: Optional status filter (empty string = all statuses)

Returns:

• []*Partition: List of partitions
• error: Database error

Example usage:

// Get all pending partitions
pending, err := tracker.ListPartitions(ctx, migration.StatusPending)

// Get all partitions regardless of status
all, err := tracker.ListPartitions(ctx, "")

func (*Tracker) ReleasePartition ¶

func (t *Tracker) ReleasePartition(ctx context.Context, partitionID, podID string) error

ReleasePartition releases a partition from in_progress back to pending.

This method is used for crash recovery: if a worker crashes while processing a partition, another worker can detect the stale claim and release it back to pending so it can be reprocessed.

The release is conditional: it only succeeds if the partition is currently claimed by the specified pod. This prevents one pod from releasing another pod's active claim.

Parameters:

• ctx: Context for the database operation
• partitionID: The partition identifier
• podID: The pod identifier that originally claimed the partition

Returns:

• error: ErrPartitionNotFound if the partition doesn't exist, ErrPartitionClaimedByAnother if claimed by a different pod, or a database error

Example usage:

// Release stale claims on startup
staleClaims, err := tracker.GetStaleClaims(ctx, 30*time.Minute)
for _, claim := range staleClaims {
    err := tracker.ReleasePartition(ctx, claim.PartitionID, *claim.ClaimedBy)
    if err != nil {
        log.Printf("Failed to release partition %s: %v\n", claim.PartitionID, err)
    }
}

func (*Tracker) UpdateStatus ¶

func (t *Tracker) UpdateStatus(ctx context.Context, partitionID string, status PartitionStatus, errorMessage *string) error

UpdateStatus updates the status of a partition.

This method is used to: - Mark a partition as completed after successful processing - Mark a partition as failed if processing encountered an error - Release a failed partition back to pending for retry

Parameters:

• ctx: Context for the database operation
• partitionID: The partition identifier
• status: The new status (completed, failed, or pending for retry)
• errorMessage: Optional error message (required for status='failed', optional otherwise)

Returns:

• error: ErrPartitionNotFound if the partition doesn't exist, or a database error

Example usage:

err := tracker.UpdateStatus(ctx, "day=2023-01-01", migration.StatusCompleted, nil)
if err != nil {
    return fmt.Errorf("failed to mark partition completed: %w", err)
}