gopipeline

package module
v0.0.0-...-8d9fe74 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Mar 7, 2026 License: MIT Imports: 12 Imported by: 0

README

gopipeline

A Go package for building resilient pipelines with journaling, debug logging, rate limiting, and progress monitoring capabilities.

Features

  • Pipeline orchestration - Chain stages with automatic error propagation
  • Journaling - Persist stage results for replay after failures
  • Debug logging - Structured logging with multiple output formats
  • Rate limiting - Built-in rate limiting with exponential backoff
  • Auto-Retry - Automatic retry for transient failures with configurable policies
  • Pause & Resume - Graceful and immediate pause support for pipeline control
  • Progress monitoring - Emit progress updates during stage execution
  • Pool processing - Worker pool for concurrent task execution
  • All execution - Run multiple independent pipelines simultaneously

Installation

go get github.com/rohmanhakim/gopipeline

Quick Start

package main

import (
    "context"
    "fmt"
    "log"

    gopipeline "github.com/rohmanhakim/gopipeline"
)

func main() {
    // Create a stage runner with journaling
    runner, err := gopipeline.NewStageRunner(
        gopipeline.NewInMemoryJournal(),
    )
    if err != nil {
        log.Fatal(err)
    }

    ctx := context.Background()

    // Process a stage
    result := gopipeline.Process(ctx, runner, "my-stage", func(progress gopipeline.ProgressFunc) gopipeline.StageResult[string] {
        return gopipeline.NewStageResult("hello world", nil)
    })

    output, err := result.Decompose()
    if err != nil {
        log.Fatal(err)
    }
    fmt.Println(output) // Output: hello world
}

Stage Chaining

Use Then() to chain stages where each stage depends on the previous one's output:

// Step 1: Fetch data
fetchStage := gopipeline.Process(ctx, runner, "fetch-data", func(progress gopipeline.ProgressFunc) gopipeline.StageResult[FetchResult] {
    return fetcher.Fetch(url)
})

// Step 2: Process data (only runs if fetch succeeded)
processStage := gopipeline.Then("process-data", fetchStage, func(f FetchResult, progress gopipeline.ProgressFunc) gopipeline.StageResult[ProcessResult] {
    return processor.Process(f.Data)
})

// Step 3: Save data (only runs if process succeeded)
saveStage := gopipeline.Then("save-data", processStage, func(p ProcessResult, progress gopipeline.ProgressFunc) gopipeline.StageResult[SaveResult] {
    return saver.Save(p.Output)
})

// Get final result
result, err := saveStage.Decompose()
Continue Despite Errors with ThenContinue

Unlike Then(), which skips execution when the previous stage fails, ThenContinue() always executes the task. The task receives both the output (may be zero value) and the error (may be nil):

// Stage 1 - might fail
fetchStage := gopipeline.Process(ctx, runner, "fetch-data", func(progress gopipeline.ProgressFunc) gopipeline.StageResult[FetchResult] {
    return fetcher.Fetch(url)
})

// Stage 2 - runs regardless of Stage 1 outcome
cleanupStage := gopipeline.ThenContinue("cleanup", fetchStage, func(result FetchResult, err gopipeline.StageError, progress gopipeline.ProgressFunc) gopipeline.StageResult[CleanupResult] {
    if err != nil {
        // Handle error case - maybe log or cleanup partial state
        log.Printf("fetch failed: %v", err)
    }
    // Always run cleanup
    return cleanup()
})

This is particularly useful with CollectAll to process partial results:

// Pool downloads with partial results preserved
downloadStage := gopipeline.Pool(ctx, runner, "download", urls,
    gopipeline.PoolConfig{Workers: 4},
    downloadTask)

// Archive whatever downloaded successfully
archiveStage := gopipeline.ThenContinue("archive", downloadStage, 
    func(results []FileResult, err gopipeline.StageError, progress gopipeline.ProgressFunc) gopipeline.StageResult[ArchiveResult] {
        if err != nil {
            if partialErr, ok := err.(*gopipeline.PartialResultsError); ok {
                fmt.Printf("Archiving %d successful downloads, %d failed\n", 
                    len(partialErr.PartialResults()), len(partialErr.Errors()))
            }
        }
        // Archive all successful results
        return archiver.Archive(results)
    })
Why Not Method Chaining?

You might wonder why StageResult[T] doesn't have a .Then() method for fluent chaining. This is a Go language limitation: methods on generic types cannot have additional type parameters.

For example, this is not valid in Go:

// This won't compile - Go doesn't allow type parameters on methods
func (r StageResult[I]) Then[O any](...) StageResult[O]

The standalone Then() function is the idiomatic workaround, providing the same functionality while working within Go's generics constraints.

Optional Stages

gopipeline provides two functions for conditional stage execution, useful for runtime-configurable workflows.

ThenOptionalEnd - Terminal Optional Stages

For optional stages at the end of a pipeline (e.g., archiving, notifications), use ThenOptionalEnd. When disabled, the stage is skipped entirely and returns nil:

// CLI flag to control optional archiving
archiveEnabled := flag.Bool("archive", false, "enable archiving")
flag.Parse()

// ... earlier stages ...

// Optional archiving at the end of pipeline
archiveStage := gopipeline.ThenOptionalEnd("archive", downloadStage, *archiveEnabled,
    func(results []FileResult, progress gopipeline.ProgressFunc) gopipeline.StageResult[any] {
        return archiver.Archive(results)
    })

result, err := archiveStage.Decompose()
if err != nil {
    log.Fatal(err)
}

// result is nil when disabled, or the archive result when enabled
if result != nil {
    fmt.Printf("Archived to: %s\n", result.(ArchiveResult).Path)
}

Use ThenOptionalEnd when:

  • The stage is at the end of the pipeline
  • You don't need the result for further processing
  • You want a simple enable/disable flag
ThenEither - Conditional Stage Branches

For conditional stages in the middle of a workflow, use ThenEither. Both branches must produce the same output type:

// CLI flag for compression mode
useCompression := flag.Bool("compress", true, "enable compression")

// Conditional transformation in the middle of pipeline
transformStage := gopipeline.ThenEither("transform", fetchStage, *useCompression,
    func(fr FetchResult, progress gopipeline.ProgressFunc) gopipeline.StageResult[TransformedResult] {
        // Branch A: With compression
        return gopipeline.NewStageResult(TransformedResult{
            Data: compress(fr.Body),
            Compressed: true,
        }, nil)
    },
    func(fr FetchResult, progress gopipeline.ProgressFunc) gopipeline.StageResult[TransformedResult] {
        // Branch B: Passthrough without compression
        return gopipeline.NewStageResult(TransformedResult{
            Data:       fr.Body,
            Compressed: false,
        }, nil)
    },
)

// Can continue chaining after ThenEither
saveStage := gopipeline.Then("save", transformStage, saveTask)

Use ThenEither when:

  • The conditional stage is in the middle of the pipeline
  • Both branches produce the same type
  • You need to continue chaining after the conditional stage
Comparison
Function Position Returns Chainable Use Case
ThenOptionalEnd End only StageResult[any] No Cleanup, archiving, notifications
ThenEither Anywhere StageResult[O] Yes Data transformation branches
Server/Service Configuration

Both functions are designed for runtime configuration without code changes:

// Load configuration from file, environment, or CLI
config := loadConfig()

// Build pipeline with optional stages
fetchStage := gopipeline.Process(ctx, runner, "fetch", fetchTask)

// Optional transform (from config)
transformStage := gopipeline.ThenEither("transform", fetchStage, config.TransformEnabled,
    transformTask,
    passthroughTask,
)

// Optional validation (from config)
validateStage := gopipeline.ThenEither("validate", transformStage, config.ValidationEnabled,
    validateTask,
    skipValidateTask,
)

// Optional archiving (from config)
archiveStage := gopipeline.ThenOptionalEnd("archive", validateStage, config.ArchiveEnabled,
    archiveTask,
)

This pattern enables dynamic workflow changes in server environments without recompilation.

Progress Monitoring

Emit progress updates during long-running stage execution. This is useful for displaying CLI progress bars, tracking download progress, or or reporting intermediate status.

Using ProgressFunc

The task function receives a ProgressFunc callback. Call it to emit progress:

result := gopipeline.Process(ctx, runner, "download-file", func(progress gopipeline.ProgressFunc) gopipeline.StageResult[FileResult] {
    resp, _ := http.Get(url)
    total := resp.ContentLength
    
    progress(0, gopipeline.FieldMap{"file": filename})
    
    var downloaded int64
    buf := make([]byte, 32*1024)
    for {
        n, err := resp.Body.Read(buf)
        downloaded += int64(n)
        
        percent := float64(downloaded) / float64(total) * 100
        progress(percent, gopipeline.FieldMap{
            "downloaded": downloaded,
            "total": total,
        })
        
        if err == io.EOF {
            break
        }
    }
    
    progress(100, nil)
    return gopipeline.NewStageResult(result, nil)
})
Custom Progress Handler

Use WithProgressHandler to handle progress events with custom logic, such as displaying a CLI progress bar:

runner, _ := gopipeline.NewStageRunner(
    gopipeline.NewInMemoryJournal(),
    gopipeline.WithProgressHandler(func(stage string, percent float64, fields gopipeline.FieldMap) {
        fmt.Printf("\r%s: %.0f%% complete", stage, percent)
        if percent >= 100 {
            fmt.Println()
        }
    }),
)
Progress with Debug Logging

When debug logging is enabled, progress events are automatically logged:

runner, _ := gopipeline.NewStageRunner(
    gopipeline.NewInMemoryJournal(),
    gopipeline.WithDebugLogger(
        gopipeline.WithFormat(gopipeline.FormatText),
        gopipeline.WithOutputFile("pipeline.log"),
    ),
)
// Progress events will be logged like:
// DEBUG Pipeline stage progress stage="download-file" percent=50 downloaded=1024 total=2048

Error Handling

StageResult[T] provides several methods for handling results:

// Traditional Go style
output, err := result.Decompose()
if err != nil {
    // handle error
}

// With fallback value
output := result.UnwrapOr(defaultValue)

// Panic on error (use with caution)
output := result.Unwrap()

Pause & Resume

Pause and resume pipeline execution for maintenance, debugging, or resource management.

Graceful Pause

Waits for the current stage to complete before pausing. New stages will block until Resume() is called:

runner.Pause(gopipeline.PauseGraceful)

// ... do maintenance work ...

runner.Resume()
Immediate Pause

Interrupts the current stage immediately via context cancellation. The running stage must check ctx.Done() to handle interruption:

runner.Pause(gopipeline.PauseImmediate)
// Current stage receives context cancellation
runner.Resume()
Status Check
if runner.IsPaused() {
    // Pipeline is paused
}
Signal-Based Control

Control the pipeline via OS signals:

sigChan := make(chan os.Signal, 1)
signal.Notify(sigChan, syscall.SIGUSR1)

go func() {
    for range sigChan {
        if runner.IsPaused() {
            runner.Resume()
        } else {
            runner.Pause(gopipeline.PauseGraceful)
        }
    }
}()
Important Note on Immediate Pause

For PauseImmediate to interrupt a running task, the task must actively check the context:

result := gopipeline.Process(ctx, runner, "long-stage", func(progress gopipeline.ProgressFunc) gopipeline.StageResult[string] {
    select {
    case <-time.After(5 * time.Second):
        return gopipeline.NewStageResult("completed", nil)
    case <-ctx.Done():
        return gopipeline.NewStageResult("", ctx.Err())
    }
})

Rate Limiting

Enable rate limiting to control request throughput:

runner, err := gopipeline.NewStageRunner(
    gopipeline.NewInMemoryJournal(),
    gopipeline.WithRateLimiter(
        gopipeline.WithBaseDelay(1*time.Second),
        gopipeline.WithJitter(100*time.Millisecond),
        gopipeline.WithMultiplier(2.0),
        gopipeline.WithMaxDuration(30*time.Second),
    ),
)

// Wait before making a request
if err := runner.Wait(ctx, "api.example.com"); err != nil {
    return err
}

// Trigger backoff on rate limit or server errors
runner.Backoff(ctx, "api.example.com")

// Reset backoff when service recovers
runner.ResetBackoff("api.example.com")
Per-Resource Delays
gopipeline.WithRateLimiter(
    gopipeline.WithBaseDelay(1*time.Second),
    gopipeline.WithResourceDelay("strict.api.com", 5*time.Second),
)

Auto-Retry

Enable automatic retry for transient failures with exponential backoff:

Basic Retry
runner, err := gopipeline.NewStageRunner(
    gopipeline.NewInMemoryJournal(),
    gopipeline.WithRetry(), // defaults: 3 attempts, fail-fast policy
)
Shared Backoff Configuration

Configure backoff settings shared by both rate limiter and retry:

runner, err := gopipeline.NewStageRunner(
    gopipeline.NewInMemoryJournal(),
    // Shared backoff config (used by both rate limiter and retry)
    gopipeline.WithBackoff(
        gopipeline.WithBackoffInitialDuration(1*time.Second),
        gopipeline.WithBackoffMultiplier(2.0),
        gopipeline.WithBackoffMaxDuration(1*time.Minute),
        gopipeline.WithBackoffJitter(100*time.Millisecond),
    ),
    gopipeline.WithRateLimiter(
        gopipeline.WithBaseDelay(2*time.Second),
    ),
    gopipeline.WithRetry(
        gopipeline.WithMaxAttempts(5),
    ),
)
Custom Retry Configuration
runner, err := gopipeline.NewStageRunner(
    gopipeline.NewInMemoryJournal(),
    gopipeline.WithRetry(
        gopipeline.WithMaxAttempts(5),
        gopipeline.WithRetryPolicy(gopipeline.RetryPolicyNever), // fail-fast
        gopipeline.WithRetryInitialDuration(500*time.Millisecond),
        gopipeline.WithRetryJitter(50*time.Millisecond),
    ),
)
Implementing Retryable Errors

For fine-grained control, implement the RetryableError interface:

import "github.com/rohmanhakim/gopipeline"

type FetchError struct {
    Message    string
    StatusCode int
    RetryAfter time.Duration // from HTTP Retry-After header
}

func (e *FetchError) Error() string {
    return fmt.Sprintf("HTTP %d: %s", e.StatusCode, e.Message)
}

// RetryPolicy determines if the error should be retried
func (e *FetchError) RetryPolicy() gopipeline.RetryPolicy {
    // Retryable: 5XX server errors, 408 Request Timeout, 429 Rate Limited
    if e.StatusCode >= 500 || e.StatusCode == 408 || e.StatusCode == 429 {
        return gopipeline.RetryPolicyAuto
    }
    // Transient: 400, 401, 403, 404 - won't be resolved by retrying
    return gopipeline.RetryPolicyNever
}

// SuggestedDelay returns server-suggested delay (e.g., from Retry-After header)
func (e *FetchError) SuggestedDelay() time.Duration {
    return e.RetryAfter
}
Retry Policies
Policy Description
RetryPolicyAuto Error will be retried automatically with exponential backoff
RetryPolicyManual Error should not be auto-retried, but eligible for manual retry
RetryPolicyNever Permanent failure, should not be retried
Behavior Matrix
Error Type RetryPolicyNever (default) RetryPolicyAuto
Standard error No retry Auto-retry
RetryableError(Auto) Auto-retry Auto-retry
RetryableError(Never) No retry No retry

Key insight: Errors implementing RetryableError always take precedence over the default policy.

Server-Suggested Delay

For HTTP 429 responses with Retry-After headers, implement DelaySuggestioner:

type RateLimitError struct {
    StatusCode int
    RetryAfter time.Duration
}

func (e *RateLimitError) SuggestedDelay() time.Duration {
    return e.RetryAfter
}

The retry backoff will use max(serverDelay, calculatedBackoff) to respect the server's suggestion.

Debug Logging

Enable structured logging for debugging:

runner, err := gopipeline.NewStageRunner(
    gopipeline.NewInMemoryJournal(),
    gopipeline.WithDebugLogger(
        gopipeline.WithFormat(gopipeline.FormatText),
        gopipeline.WithOutputFile("pipeline.log"),
    ),
)
Log Formats
  • FormatJSON - JSON structured logs
  • FormatText - Human-readable text
  • FormatLogfmt - logfmt style
  • FormatLogstash - Logstash-compatible JSON (default)

Journaling

Journals persist stage results for replay after failures. When a stage is rerun, the journal is checked first. If a successful result exists, it's returned without re-executing the task.

Failure Tracking

The journal now tracks both successful and failed stages, enabling:

  1. Crash Recovery: If a pipeline crashes, restart with the same journal file to:

    • Replay successful stages from the journal (no re-execution)
    • Re-execute failed stages automatically

  2. Manual Retry: List failed stages and retry them after fixing issues

// Create journal
journal, err := gopipeline.NewGobFileJournal("pipeline_state.gob")

runner, _ := gopipeline.NewStageRunner(journal)

// Run pipeline - failures are automatically recorded
result := gopipeline.Process(ctx, runner, "fetch-data", func(progress gopipeline.ProgressFunc) gopipeline.StageResult[Data] {
    return fetchData(url)
})

// Later: Check for failures
failures := gopipeline.ListFailures(runner)
for _, name := range failures {
    record, _ := gopipeline.GetStageRecord(runner, name)
    fmt.Printf("Failed: %s, Attempts: %d, Error: %v\n", name, record.Attempts, record.Error)
}

// Clear failure record before retrying
gopipeline.ClearFailure(runner, "fetch-data")

// Re-run the pipeline - failed stage will be re-executed
result = gopipeline.Process(ctx, runner, "fetch-data", func(progress gopipeline.ProgressFunc) gopipeline.StageResult[Data] {
    return fetchData(url)
})
StageRecord Structure

Each stage is stored as a StageRecord:

type StageRecord struct {
    Status    StageStatus  // "success" or "failed"
    Output    any          // Result (for successful stages)
    Input     any          // Input data (for retry capability)
    Error     any          // Error information (for failed stages)
    Timestamp time.Time    // When the stage was executed
    Attempts  int          // How many times the stage has been attempted
}
InMemoryJournal

For testing or non-critical pipelines where persistence is not required:

journal := gopipeline.NewInMemoryJournal()
GobFileJournal (Persistent)

For durable persistence across pipeline restarts with full type safety:

// Register all stage result types when creating the journal.
// This is required for gob to properly serialize/deserialize your data.
journal, err := gopipeline.NewGobFileJournal("pipeline_state.gob",
    FetchResult{},      // register FetchResult
    ExtractionResult{}, // register ExtractionResult
    WriteResult{},      // register WriteResult
)
if err != nil {
    log.Fatal(err)
}
runner, _ := gopipeline.NewStageRunner(journal)

Why Gob?

  • Type-safe: Preserves type information during serialization
  • Required Registration: Types must be passed to NewGobFileJournal for proper serialization
  • Efficient: Binary format, smaller and faster than JSON

⚠️ Important: Type Registration Required

All types used as stage outputs must be passed to NewGobFileJournal. This ensures:

  • New journals can properly serialize your data
  • Existing journals can be loaded and replayed correctly
// ✅ CORRECT - pass all result types when creating journal
journal, err := gopipeline.NewGobFileJournal("pipeline_state.gob",
    FetchResult{},
    ExtractionResult{},
    WriteResult{},
)

⚠️ Important: Exported Fields Required

Gob encoding can only serialize exported fields (fields starting with an uppercase letter). If your struct has unexported fields, the journal Save() will return an error and the pipeline will fail.

// ❌ WRONG - unexported fields cannot be serialized
type FetchResult struct {
    url       string    // unexported - won't be saved!
    body      []byte    // unexported - won't be saved!
    fetchedAt time.Time // unexported - won't be saved!
}

// ✅ CORRECT - exported fields are serializable
type FetchResult struct {
    URL       string    // exported - will be saved
    Body      []byte    // exported - will be saved
    FetchedAt time.Time // exported - will be saved
}

⚠️ Important: Avoid Non-Serializable Types

Some Go types cannot be serialized by gob, such as url.URL which has unexported fields. Use serializable alternatives:

// ❌ WRONG - url.URL has unexported fields
type FetchResult struct {
    URL url.URL // gob cannot serialize this!
}

// ✅ CORRECT - use string instead
type FetchResult struct {
    URL string // gob can serialize this
}

When using GobFileJournal, ensure all fields in your result structs are exported and serializable. The pipeline will return an error during Save() if serialization fails, allowing you to catch this issue before deploying to production.

Pool Failure Tracking

Pool processing supports two failure tracking modes:

// Pool-level tracking (default): entire pool as one failure unit
runner, _ := gopipeline.NewStageRunner(journal,
    gopipeline.WithPoolFailureMode(gopipeline.PoolFailurePoolLevel),
)

// Individual tracking: each pool item tracked separately
runner, _ := gopipeline.NewStageRunner(journal,
    gopipeline.WithPoolFailureMode(gopipeline.PoolFailureIndividual),
)
Pool-Level Tracking (Default)

When a pool fails, the entire pool's failure is recorded and can be inspected:

record, ok := gopipeline.GetStageRecord(runner, "download")
if ok && record.Status == gopipeline.StatusFailed {
    fmt.Printf("Pool failed: %v\n", record.Error)
}
Individual Tracking

With PoolFailureIndividual, each pool item is tracked separately. This enables fine-grained retry control:

runner, _ := gopipeline.NewStageRunner(journal,
    gopipeline.WithPoolFailureMode(gopipeline.PoolFailureIndividual),
)

// Run pool with 5 items (items 2 and 4 fail)
gopipeline.Pool(ctx, runner, "download", urls,
    gopipeline.PoolConfig{Workers: 4},
    downloadTask)

// List failures - returns individual failed items
failures := gopipeline.ListFailures(runner)
// ["download [2]", "download [4]"]

// Inspect specific failure with input data
record, _ := gopipeline.GetStageRecord(runner, "download [2]")
fmt.Printf("Failed URL: %v\n", record.Input)  // Original input preserved

// Clear specific failure and re-run
gopipeline.ClearFailure(runner, "download [2]")
gopipeline.Pool(ctx, runner, "download", urls, ...) // Only item 2 retries

When re-running a pool without calling ClearFailure():

  • Successful items: Replayed from journal (skipped)
  • Failed items: Automatically retried

This matches the existing single-stage behavior, giving users control over retry granularity.

Custom Journal

Implement the Journal interface for other storage backends (databases, key-value stores, etc.):

type MyJournal struct{}

func (j *MyJournal) Save(name string, record gopipeline.StageRecord) {
    // Persist to database/file
}

func (j *MyJournal) Load(name string) (gopipeline.StageRecord, bool) {
    // Load from database/file
    return gopipeline.StageRecord{}, false
}

func (j *MyJournal) ListFailures() []string {
    // Return names of failed stages
    return nil
}

func (j *MyJournal) ListAll() []string {
    // Return all stage names
    return nil
}

func (j *MyJournal) Clear(name string) {
    // Remove specific stage
}

func (j *MyJournal) ClearAll() {
    // Remove all stages
}

⚠️ Warning: Type Safety in Custom Journals

When implementing your own Journal, ensure that the Output field in StageRecord preserves type information during serialization. Standard JSON encoding loses type information (structs become map[string]interface{}), which causes runtime panics when the pipeline attempts type assertions on restored results.

Solution: Type Registry Pattern

If you need JSON format with type safety, implement a type registry:

import (
    "encoding/json"
    "reflect"
)

// 1. Define a wrapper for type-safe serialization
type TypedOutput struct {
    Type string          `json:"type"`
    Data json.RawMessage `json:"data"`
}

// 2. Create a type registry
var typeRegistry = map[string]func() any{
    "FetchResult":      func() any { return &FetchResult{} },
    "ExtractionResult": func() any { return &ExtractionResult{} },
    "WriteResult":      func() any { return &WriteResult{} },
}

// 3. Register types before using the journal
func RegisterType(name string, prototype any) {
    typeRegistry[name] = func() any { return prototype }
}

// 4. Custom journal with type registry
type TypedJSONJournal struct {
    filePath string
    stages   map[string]StageRecord
}

func (j *TypedJSONJournal) Save(name string, record StageRecord) {
    // Wrap output with type name before serializing
    if record.Output != nil {
        typeName := reflect.TypeOf(record.Output).Name()
        data, _ := json.Marshal(record.Output)
        // Store with type wrapper in your persistence layer
        // The actual implementation depends on your storage backend
        _ = TypedOutput{Type: typeName, Data: data}
    }
    // ... persist to your storage backend
}

func (j *TypedJSONJournal) Load(name string) (StageRecord, bool) {
    // Load and reconstruct the correct type
    record := StageRecord{} // load from your storage backend
    if typed, ok := record.Output.(TypedOutput); ok {
        if factory, exists := typeRegistry[typed.Type]; exists {
            target := factory()
            json.Unmarshal(typed.Data, target)
            record.Output = target
        }
    }
    return record, true
}

// Implement remaining Journal methods...

Recommendation: For most use cases, prefer GobFileJournal which handles type preservation automatically without additional complexity.

Pool Processing

Process multiple items concurrently using a worker pool. Perfect for I/O-bound operations like downloading files, fetching APIs, or batch processing.

Basic Usage
urls := []string{"url1", "url2", "url3"}

result := gopipeline.Pool(ctx, runner, "download", urls,
    gopipeline.PoolConfig{Workers: 4},
    func(ctx context.Context, runner *gopipeline.StageRunner, idx int, url string, stageName string) gopipeline.StageResult[FileResult] {
        return gopipeline.Process(ctx, runner, stageName, func(progress gopipeline.ProgressFunc) gopipeline.StageResult[FileResult] {
            return downloadFile(url)
        })
    })

results, err := result.Decompose()
How It Works
  1. Worker Pool: Spawns N workers that process items from a shared input channel
  2. Stage Naming: Each item gets a unique stage name: "baseName [1]", "baseName [2]", etc.
  3. Journal Integration: Each stage's result is journaled independently for replay
  4. Pause/Resume: Workers respect pipeline pause state
Error Handling Strategies

Configure how pool processing handles errors:

runner, _ := gopipeline.NewStageRunner(
    gopipeline.NewInMemoryJournal(),
    // Fail-fast (default): Stop all workers on first error
    gopipeline.WithErrorStrategy(gopipeline.FailFast),
)
Strategy Behavior Results on Error
FailFast (default) Stop all workers immediately when one fails Discarded
CollectAll Continue processing all items, collect all errors Preserved 
// Collect all errors - useful for batch processing where you want to know all failures
runner, _ := gopipeline.NewStageRunner(
    gopipeline.NewInMemoryJournal(),
    gopipeline.WithErrorStrategy(gopipeline.CollectAll),
)
Partial Results with CollectAll

When using CollectAll, partial results are preserved alongside errors. The error returned is a PartialResultsError that provides access to both:

result := gopipeline.Pool(ctx, runner, "download", urls,
    gopipeline.PoolConfig{Workers: 4},
    downloadTask)

outputs, err := result.Decompose()
if err != nil {
    // Check if it's a partial results error
    if partialErr, ok := err.(*gopipeline.PartialResultsError); ok {
        // Access successful results
        partial := partialErr.PartialResults()
        fmt.Printf("Succeeded: %d, Failed: %d\n", len(partial), len(partialErr.Errors()))
        
        // Access all errors
        for _, e := range partialErr.Errors() {
            fmt.Printf("Error: %v\n", e)
        }
    }
    return err
}
// outputs slice contains all successful results (zero values for failed indices)
Progress Aggregation

Track overall progress across all pool workers:

runner, _ := gopipeline.NewStageRunner(
    gopipeline.NewInMemoryJournal(),
    gopipeline.WithMultiTaskProgressHandler(func(stage string, completed int, total int, percent float64, fields gopipeline.FieldMap) {
        fmt.Printf("\r%s: %d/%d complete (%.1f%%)", stage, completed, total, percent)
        if completed >= total {
            fmt.Println()
        }
    }),
)
Chaining Stages Within Pool

Each pool task can chain multiple stages:

gopipeline.Pool(ctx, runner, "process-items", items,
    gopipeline.PoolConfig{Workers: 4},
    func(ctx context.Context, runner *gopipeline.StageRunner, idx int, item Item, stageName string) gopipeline.StageResult[Result] {
        // Step 1: Fetch
        fetchStage := gopipeline.Process(ctx, runner, stageName+" fetch", func(progress gopipeline.ProgressFunc) gopipeline.StageResult[Data] {
            return fetchData(item.URL)
        })

        // Step 2: Process (only runs if fetch succeeded)
        processStage := gopipeline.Then(stageName+" process", fetchStage, func(data Data, progress gopipeline.ProgressFunc) gopipeline.StageResult[Result] {
            return processData(data)
        })

        return processStage
    })
Rate Limiting with Pool

Rate limiting works with pool processing - each worker respects the rate limiter:

gopipeline.Pool(ctx, runner, "download", urls,
    gopipeline.PoolConfig{Workers: 4},
    func(ctx context.Context, runner *gopipeline.StageRunner, idx int, url string, stageName string) gopipeline.StageResult[FileResult] {
        // Wait for rate limiter before processing
        if err := runner.Wait(ctx, "api.example.com"); err != nil {
            return gopipeline.NewStageResult(FileResult{}, err)
        }
        return gopipeline.Process(ctx, runner, stageName, func(progress gopipeline.ProgressFunc) gopipeline.StageResult[FileResult] {
            return downloadFile(url)
        })
    })

All Execution

Run multiple independent pipelines simultaneously. Unlike Pool, which processes homogeneous items with a worker pool, All runs a fixed set of different pipelines at the same time, with one goroutine per task.

When to Use All vs Pool
Feature Pool All
Data Homogeneous (same type) Heterogeneous (different types)
Operations Same operation on many items Different operations
Worker Pool Yes, configurable No (1 goroutine per task)
Scalability Designed for many items Designed for few pipelines
Use Case Batch processing Multi-pipeline orchestration
Basic Usage
// Run 3 completely different pipelines concurrently
results := gopipeline.All(ctx, runner,
    func() gopipeline.StageResult[Result] {
        // Pipeline A: Fetch from API
        return gopipeline.Process(ctx, runner, "fetch-api", func(progress gopipeline.ProgressFunc) gopipeline.StageResult[Result] {
            return fetchFromAPI()
        })
    },
    func() gopipeline.StageResult[Result] {
        // Pipeline B: Parse local files
        return gopipeline.Process(ctx, runner, "parse-local", func(progress gopipeline.ProgressFunc) gopipeline.StageResult[Result] {
            return parseFiles()
        })
    },
    func() gopipeline.StageResult[Result] {
        // Pipeline C: Write to database
        return gopipeline.Process(ctx, runner, "write-db", func(progress gopipeline.ProgressFunc) gopipeline.StageResult[Result] {
            return writeToDB()
        })
    },
)

outputs, err := results.Decompose()
Different Pipeline Lengths

Each task can have a completely different pipeline structure:

results := gopipeline.All(ctx, runner,
    func() gopipeline.StageResult[string] {
        // Single-stage pipeline
        return gopipeline.Process(ctx, runner, "quick-check", checkHealth)
    },
    func() gopipeline.StageResult[string] {
        // Multi-stage pipeline
        fetchStage := gopipeline.Process(ctx, runner, "fetch", fetchData)
        parseStage := gopipeline.Then("parse", fetchStage, parseData)
        return gopipeline.Then("transform", parseStage, transformData)
    },
)
Combining All with Pool

Run multiple batch operations simultaneously:

results := gopipeline.All(ctx, runner,
    func() gopipeline.StageResult[any] {
        // Pipeline A: Download wallpapers (pool batch)
        return gopipeline.Pool(ctx, runner, "wallpapers", wallpaperURLs,
            gopipeline.PoolConfig{Workers: 4},
            downloadTask).UnwrapOr(nil)
    },
    func() gopipeline.StageResult[any] {
        // Pipeline B: Download thumbnails (pool batch)
        return gopipeline.Pool(ctx, runner, "thumbnails", thumbnailURLs,
            gopipeline.PoolConfig{Workers: 2},
            downloadTask).UnwrapOr(nil)
    },
)
Error Handling

All uses the same error handling strategies as Pool:

runner, _ := gopipeline.NewStageRunner(
    gopipeline.NewInMemoryJournal(),
    // Collect all errors, preserve partial results
    gopipeline.WithErrorStrategy(gopipeline.CollectAll),
)

results := gopipeline.All(ctx, runner,
    func() gopipeline.StageResult[Result] { return task1() },
    func() gopipeline.StageResult[Result] { return task2() }, // fails
    func() gopipeline.StageResult[Result] { return task3() },
)

outputs, err := results.Decompose()
if err != nil {
    if partialErr, ok := err.(*gopipeline.PartialResultsError); ok {
        // Access successful results
        fmt.Printf("Succeeded: %d, Failed: %d\n",
            len(partialErr.PartialResults()), len(partialErr.Errors()))
    }
}
Progress Tracking

Track completion across all tasks:

runner, _ := gopipeline.NewStageRunner(
    gopipeline.NewInMemoryJournal(),
    gopipeline.WithMultiTaskProgressHandler(func(stage string, completed, total int, percent float64, fields gopipeline.FieldMap) {
        fmt.Printf("\r%s: %d/%d tasks complete (%.1f%%)", stage, completed, total, percent)
    }),
)

Examples

License

MIT

Documentation

Index

Constants

This section is empty.

Variables

View Source 
var NewRetryError = retrier.NewRetryError

NewRetryError creates a new RetryError with the given cause, message, policy, and wrapped error.

This is a re-export of retrier.NewRetryError for convenience, so users don't need to import the retrier package separately.

Functions

func ClearAllFailures 

func ClearAllFailures(runner *StageRunner)

ClearAllFailures removes all failure records from the journal. This allows all stages to be re-executed with fresh attempt counters.

Example usage: 

gopipeline.ClearAllFailures(runner)

func ClearFailure 

func ClearFailure(runner *StageRunner, name string)

ClearFailure removes a failure record for a stage. This allows the stage to be re-executed with a fresh attempt counter.

Example usage: 

// Clear a failure record before retrying
gopipeline.ClearFailure(runner, "fetch-data")

func ListFailures 

func ListFailures(runner *StageRunner) []string

ListFailures returns the names of all stages that have failed. This is a convenience method to check which stages need retry.

Example usage: 

failures := gopipeline.ListFailures(runner)
for _, name := range failures {
    fmt.Printf("Failed stage: %s\n", name)
}

func RegisterType 

func RegisterType[T any]()

RegisterType registers a type with gob for serialization in journals. Call this function before creating a GobFileJournal if you need to replay from an existing journal file. Types used as stage outputs must be registered before the journal attempts to load them.

For new pipelines (no existing journal), types are automatically registered when Process() is called, so explicit registration is not required.

This function also automatically registers the slice variant of the type (e.g., []T) to support Pool operations that output slices.

Example usage: 

func init() {
    gopipeline.RegisterType[FetchResult]()
    gopipeline.RegisterType[WriteResult]()
}

Types

type AllTask 

type AllTask[T any] func() StageResult[T]

AllTask is a function that returns a StageResult. Used for running heterogeneous stage chains concurrently. Each task can represent a completely different pipeline with its own stages, inputs, and outputs.

type BackoffOption 

type BackoffOption func() ratelimiter.BackoffOptions

BackoffOption is a functional option for backoff configuration.

func BackoffWithServerDelay 

func BackoffWithServerDelay(delay time.Duration) BackoffOption

BackoffWithServerDelay sets a server-suggested delay for backoff. Use this when the server provides a Retry-After header.

type DebugLogger 

type DebugLogger struct {
	// contains filtered or unexported fields
}

DebugLogger provides domain-specific logging methods for the crawler. It wraps dlog.DebugLogger and translates domain types to FieldMap.

func NewDebugLogger 

func NewDebugLogger(logger dlog.DebugLogger) *DebugLogger

NewDebugLogger creates a new DebugLogger wrapping a dlog.DebugLogger. If logger is nil, returns a logger wrapping dlog.NewNoOpLogger().

func (*DebugLogger) Close 

func (d *DebugLogger) Close() error

Close flushes any buffered output and closes file handles.

func (*DebugLogger) Enabled 

func (d *DebugLogger) Enabled() bool

Enabled returns true if debug logging is enabled.

func (*DebugLogger) LogBackoff 

func (d *DebugLogger) LogBackoff(ctx context.Context, resource string, count int, delay time.Duration, attrs ...any)

LogBackoff logs a rate limiter backoff event. This implements the ratelimiter.DebugLogger interface.

func (*DebugLogger) LogDebug 

func (d *DebugLogger) LogDebug(ctx context.Context, msg string, fields ...dlog.FieldMap)

LogDebug logs a debug-level message with optional fields.

func (*DebugLogger) LogError 

func (d *DebugLogger) LogError(ctx context.Context, msg string, err error, fields ...dlog.FieldMap)

LogError logs an error-level message with the error and optional fields.

func (*DebugLogger) LogInfo 

func (d *DebugLogger) LogInfo(ctx context.Context, msg string, fields ...dlog.FieldMap)

LogInfo logs an info-level message with optional fields.

func (*DebugLogger) LogRateLimit 

func (d *DebugLogger) LogRateLimit(ctx context.Context, resource string, delay time.Duration, reason ratelimiter.RateLimitReason, attrs ...any)

LogRateLimit logs a rate limit wait event. This implements the ratelimiter.DebugLogger interface.

func (*DebugLogger) LogRetry 

func (d *DebugLogger) LogRetry(ctx context.Context, attempt, maxAttempts int, backoff time.Duration, err error, attrs ...any)

LogRetry logs a retry attempt event. This implements the retrier.DebugLogger interface.

func (*DebugLogger) LogStage 

func (d *DebugLogger) LogStage(ctx context.Context, stage string, event StageEvent)

LogStage logs a pipeline stage event with structured fields.

func (*DebugLogger) LogStep 

func (d *DebugLogger) LogStep(ctx context.Context, stage string, step string, fields FieldMap)

LogStep logs a granular step within a pipeline stage.

func (*DebugLogger) LogWarn 

func (d *DebugLogger) LogWarn(ctx context.Context, msg string, fields ...dlog.FieldMap)

LogWarn logs a warning-level message with optional fields.

type DelaySuggestioner 

type DelaySuggestioner = retrier.DelaySuggestioner

DelaySuggestioner is an interface for errors that can suggest a delay duration. Implement this interface on your error types to provide server-suggested delays (e.g., from HTTP Retry-After headers).

This is a re-export of retrier.DelaySuggestioner for convenience, so users don't need to import the retrier package separately.

type ErrorStrategy 

type ErrorStrategy int

ErrorStrategy defines how multi-task execution handles errors. 

const (
	// FailFast stops all tasks immediately when one fails.
	// This is the default strategy and matches the behavior of sequential pipelines.
	FailFast ErrorStrategy = iota

	// CollectAll continues processing all items even when some fail.
	// All errors are collected and returned as a combined error.
	CollectAll ErrorStrategy = iota
)

type FieldMap 

type FieldMap map[string]any

FieldMap is a map of structured field names to values.

type Format 

type Format string

Format represents the output format for debug logging. 

const (
	// FormatJSON outputs logs in JSON format (Logstash/Elasticsearch compatible).
	FormatJSON Format = "json"
	// FormatText outputs logs in human-readable text format.
	FormatText Format = "text"
	// FormatLogfmt outputs logs in logfmt format (key=value pairs).
	FormatLogfmt Format = "logfmt"
	// FormatLogstash outputs logs in Logstash/Elasticsearch compatible JSON format.
	FormatLogstash Format = "logstash"
)

type GobFileJournal 

type GobFileJournal struct {
	// contains filtered or unexported fields
}

GobFileJournal is a persistent journal that stores stage results using gob encoding. It provides durability across pipeline restarts, allowing stages to be replayed without re-execution. Supports both success and failure tracking.

Unlike JSON, gob encoding preserves type information, ensuring that stored outputs can be correctly type-asserted when loaded back.

Type Registration: All types that will be stored as stage outputs must be registered with gob. Pass sample values of each type to NewGobFileJournal when creating the journal. This ensures types are registered before loading existing journal data.

func NewGobFileJournal 

func NewGobFileJournal(filePath string, types ...any) (*GobFileJournal, error)

NewGobFileJournal creates a new GobFileJournal that persists to the specified file. If the file exists, it loads existing stage results. If not, it creates a new empty journal.

The types parameter accepts sample values of all types that will be stored as stage outputs. This is required for gob to properly serialize and deserialize your data, especially when replaying from an existing journal file.

Example usage: 

journal, err := gopipeline.NewGobFileJournal("pipeline_state.gob",
    FetchResult{},      // register FetchResult
    ExtractionResult{}, // register ExtractionResult
    WriteResult{},      // register WriteResult
)
if err != nil {
    log.Fatal(err)
}
runner, _ := gopipeline.NewStageRunner(journal)

func (*GobFileJournal) Clear 

func (j *GobFileJournal) Clear(name string) error

Clear removes a specific stage record from memory and persists the change. Returns an error if the persistence fails.

func (*GobFileJournal) ClearAll 

func (j *GobFileJournal) ClearAll()

ClearAll removes all stage records from memory and deletes the gob file.

func (*GobFileJournal) FilePath 

func (j *GobFileJournal) FilePath() string

FilePath returns the path to the gob file.

func (*GobFileJournal) ListAll 

func (j *GobFileJournal) ListAll() []string

ListAll returns names of all stored stages.

func (*GobFileJournal) ListFailures 

func (j *GobFileJournal) ListFailures() []string

ListFailures returns names of all stages with StatusFailed.

func (*GobFileJournal) Load 

func (j *GobFileJournal) Load(name string) (StageRecord, bool)

Load retrieves a previously saved stage record by name. Returns the record and true if found, empty record and false otherwise.

func (*GobFileJournal) Save 

func (j *GobFileJournal) Save(name string, record StageRecord) error

Save stores a stage record with the given name and persists it to the gob file. The save is atomic - it writes to a temp file first, then renames to avoid corruption. Returns an error if the persistence fails.

type InMemoryJournal 

type InMemoryJournal struct {
	// contains filtered or unexported fields
}

InMemoryJournal is an in-memory implementation of Journal. Useful for testing or non-critical pipelines where persistence is not required.

func NewInMemoryJournal 

func NewInMemoryJournal() *InMemoryJournal

NewInMemoryJournal creates a new InMemoryJournal

func (*InMemoryJournal) Clear 

func (j *InMemoryJournal) Clear(name string) error

Clear removes a specific stage record

func (*InMemoryJournal) ClearAll 

func (j *InMemoryJournal) ClearAll()

ClearAll removes all stage records

func (*InMemoryJournal) ListAll 

func (j *InMemoryJournal) ListAll() []string

ListAll returns names of all stored stages

func (*InMemoryJournal) ListFailures 

func (j *InMemoryJournal) ListFailures() []string

ListFailures returns names of all failed stages

func (*InMemoryJournal) Load 

func (j *InMemoryJournal) Load(name string) (StageRecord, bool)

Load retrieves a stage record by name

func (*InMemoryJournal) Save 

func (j *InMemoryJournal) Save(name string, record StageRecord) error

Save stores a stage record

type Journal 

type Journal interface {
	// Save stores a stage record with the given name.
	// Returns an error if the persistence fails.
	Save(name string, record StageRecord) error

	// Load retrieves a stage record by name
	Load(name string) (StageRecord, bool)

	// ListFailures returns names of all stages with StatusFailed
	ListFailures() []string

	// ListAll returns names of all stored stages
	ListAll() []string

	// Clear removes a specific stage record.
	// Returns an error if the persistence fails.
	Clear(name string) error

	// ClearAll removes all stage records
	ClearAll()
}

Journal interface for persisting stage results. Enhanced to support both success and failure tracking.

type LoggerOption 

type LoggerOption func(*loggerConfig)

LoggerOption is a functional option for configuring the debug logger.

func WithFormat 

func WithFormat(f Format) LoggerOption

WithFormat sets the output format for the debug logger. Default is FormatLogstash.

func WithLoggerFields 

func WithLoggerFields(fields dlog.FieldMap) LoggerOption

WithLoggerFields sets pre-populated fields for all log entries. These fields will be included in every log message from this logger.

func WithMinLevel 

func WithMinLevel(level dlog.Level) LoggerOption

WithMinLevel sets the minimum log level. Default is LevelDebug.

func WithOutputFile 

func WithOutputFile(path string) LoggerOption

WithOutputFile enables file output in addition to stdout. If not specified, logs are written to stdout only.

func WithSyncInterval 

func WithSyncInterval(interval time.Duration) LoggerOption

WithSyncInterval sets the flush interval for SyncPeriodic mode. Only used when SyncMode is SyncPeriodic.

func WithSyncMode 

func WithSyncMode(mode dlog.SyncMode) LoggerOption

WithSyncMode sets the sync mode for file output. Default is SyncImmediate for maximum durability.

type MultiTaskProgressHandler 

type MultiTaskProgressHandler func(stage string, completed int, total int, percent float64, fields FieldMap)

MultiTaskProgressHandler is a callback function for handling aggregated progress across multiple concurrent tasks.

type PartialResultsError 

type PartialResultsError struct {
	// contains filtered or unexported fields
}

PartialResultsError is an error type that carries partial results from pool processing. When using CollectAll strategy, this error is returned when some items fail, but successful results are preserved and accessible via PartialResults().

func (*PartialResultsError) Error 

func (e *PartialResultsError) Error() string

Error implements the error interface

func (*PartialResultsError) Errors 

func (e *PartialResultsError) Errors() []error

Errors returns all collected errors

func (*PartialResultsError) PartialResults 

func (e *PartialResultsError) PartialResults() []any

PartialResults returns the successful results from pool processing. The returned slice contains items of type any that must be type-asserted by the caller. This is useful when you want to process successful results even when some items failed.

type PauseMode 

type PauseMode int

PauseMode defines how the pipeline should pause. 

const (
	// PauseGraceful waits for the current stage to complete before pausing.
	// New stages will block until Resume() is called.
	PauseGraceful PauseMode = iota

	// PauseImmediate interrupts the current stage immediately.
	// The running stage will receive a context cancellation signal.
	PauseImmediate PauseMode = iota
)

type PoolConfig 

type PoolConfig struct {
	// Workers is the number of concurrent workers.
	// If 0 or negative, defaults to 1 (sequential).
	Workers int
}

PoolConfig holds configuration for pool processing.

type PoolFailureMode 

type PoolFailureMode int

PoolFailureMode controls how pool failures are tracked in the journal. 

const (
	// PoolFailurePoolLevel tracks the entire pool as one failure unit.
	// This is the default and simplest mode.
	PoolFailurePoolLevel PoolFailureMode = iota

	// PoolFailureIndividual tracks each pool item separately.
	// Each failed item gets its own journal entry (e.g., "download [2]", "download [4]").
	// Failed items are automatically retried when the pool is re-run.
	// Input data is preserved in the journal for retry capability.
	PoolFailureIndividual
)

type PoolRecord 

type PoolRecord struct {
	// Status indicates whether the pool completed successfully
	Status StageStatus `json:"status"`

	// Outputs contains the results from all pool items
	Outputs []any `json:"outputs,omitempty"`

	// Inputs contains the original input items
	Inputs []any `json:"inputs,omitempty"`

	// FailedIndices contains the indices of failed items
	FailedIndices []int `json:"failedIndices,omitempty"`

	// Errors contains error information for failed items
	Errors []any `json:"errors,omitempty"`

	// Timestamp when the pool was executed
	Timestamp time.Time `json:"timestamp"`

	// Attempts counts how many times this pool has been attempted
	Attempts int `json:"attempts"`
}

PoolRecord wraps pool results with metadata for journaling. This tracks pool execution results including partial failures.

type PoolTask 

type PoolTask[I any, O any] func(ctx context.Context, runner *StageRunner, index int, input I, stageName string) StageResult[O]

PoolTask is the function signature for pool task execution. It receives:

  • ctx: the context for cancellation
  • runner: the StageRunner for executing stages
  • index: the 0-based index of the input item
  • input: the input item to process
  • stageName: the auto-generated stage name (format: "baseName [index+1]")

The task should use the provided stageName when calling Process or Then to ensure unique journal keys.

type ProgressFunc 

type ProgressFunc func(percent float64, fields FieldMap)

ProgressFunc is a callback function for emitting stage progress. Percent should be 0-100, fields can contain additional structured data. If debug logging is disabled, calls to ProgressFunc are no-ops.

type ProgressHandler 

type ProgressHandler func(stage string, percent float64, fields FieldMap)

ProgressHandler is a callback function for handling progress events. This allows custom handling of progress updates, such as displaying a CLI progress bar.

type RateLimiterOption 

type RateLimiterOption func(*rateLimiterConfig)

RateLimiterOption is a functional option for configuring rate limiting.

func WithBaseDelay 

func WithBaseDelay(d time.Duration) RateLimiterOption

WithBaseDelay sets the global minimum delay between requests. All requests will wait at least this long before executing. Default is 0 (no base delay).

func WithInitialDuration 

func WithInitialDuration(d time.Duration) RateLimiterOption

WithInitialDuration sets the initial backoff duration. This is used when Backoff() is first called for a resource. Default is 1 second.

func WithJitter 

func WithJitter(d time.Duration) RateLimiterOption

WithJitter sets the maximum random delay added to rate limits. Jitter helps avoid thundering herd problems in distributed systems. Default is 0 (no jitter).

func WithMaxDuration 

func WithMaxDuration(d time.Duration) RateLimiterOption

WithMaxDuration sets the maximum backoff duration cap. Backoff delays will never exceed this value. Default is 1 minute.

func WithMultiplier 

func WithMultiplier(m float64) RateLimiterOption

WithMultiplier sets the exponential backoff multiplier. Each successive Backoff() call multiplies the delay by this factor. Default is 2.0.

func WithRateLimiterLogAttrs 

func WithRateLimiterLogAttrs(attrs ...any) RateLimiterOption

WithRateLimiterLogAttrs sets additional attributes for rate limiter logging. These attributes will be included in all rate limiter log messages.

func WithResourceDelay 

func WithResourceDelay(resource string, delay time.Duration) RateLimiterOption

WithResourceDelay sets a specific delay for a given resource. This overrides the base delay for that resource. Can be called multiple times for different resources.

type RetryError 

type RetryError = retrier.RetryError

RetryError represents a retry-related error with cause and policy information.

This is a re-export of retrier.RetryError for convenience, so users don't need to import the retrier package separately.

type RetryErrorCause 

type RetryErrorCause = retrier.RetryErrorCause

RetryErrorCause represents the cause of a retry error.

This is a re-export of retrier.RetryErrorCause for convenience, so users don't need to import the retrier package separately. 

const (
	// ErrZeroAttempt indicates that MaxAttempts was set to 0.
	ErrZeroAttempt RetryErrorCause = retrier.ErrZeroAttempt

	// ErrExhaustedAttempts indicates that all retry attempts were exhausted.
	ErrExhaustedAttempts RetryErrorCause = retrier.ErrExhaustedAttempts

	// ErrContextCancelled indicates that the context was cancelled during retry.
	ErrContextCancelled RetryErrorCause = retrier.ErrContextCancelled
)

type RetryOption 

type RetryOption func(*retryConfig)

RetryOption is a functional option for configuring auto-retry.

func WithMaxAttempts 

func WithMaxAttempts(n int) RetryOption

WithMaxAttempts sets the maximum number of retry attempts. Default is 3.

func WithRetryInitialDuration 

func WithRetryInitialDuration(d time.Duration) RetryOption

WithRetryInitialDuration sets the initial retry backoff duration. This overrides the shared backoff initial duration for retry only. Default is 1 second.

func WithRetryJitter 

func WithRetryJitter(d time.Duration) RetryOption

WithRetryJitter sets the maximum random delay added to retry backoff. This overrides the shared backoff jitter for retry only. Jitter helps avoid thundering herd problems in distributed systems. Default is 0 (no jitter).

func WithRetryLogAttrs 

func WithRetryLogAttrs(attrs ...any) RetryOption

WithRetryLogAttrs sets additional attributes for retry logging. These attributes will be included in all retry log messages.

func WithRetryMaxDuration 

func WithRetryMaxDuration(d time.Duration) RetryOption

WithRetryMaxDuration sets the maximum retry backoff duration cap. This overrides the shared backoff max duration for retry only. Retry backoff delays will never exceed this value. Default is 1 minute.

func WithRetryMultiplier 

func WithRetryMultiplier(m float64) RetryOption

WithRetryMultiplier sets the exponential backoff multiplier for retry. This overrides the shared backoff multiplier for retry only. Each successive retry multiplies the delay by this factor. Default is 2.0.

func WithRetryPolicy 

func WithRetryPolicy(p RetryPolicy) RetryOption

WithRetryPolicy sets the default retry policy for standard errors. - RetryPolicyAuto: All standard errors will be retried automatically - RetryPolicyNever: Only errors implementing RetryableError with RetryPolicyAuto will be retried (fail-fast) Default is RetryPolicyNever (fail-fast).

type RetryPolicy 

type RetryPolicy = retrier.RetryPolicy

RetryPolicy defines automatic retry behavior. This controls whether Retry() will attempt exponential backoff.

This is a re-export of retrier.RetryPolicy for convenience, so users don't need to import the retrier package separately. 

const (
	// RetryPolicyAuto indicates the error should be retried automatically
	// with exponential backoff.
	RetryPolicyAuto RetryPolicy = retrier.RetryPolicyAuto

	// RetryPolicyManual indicates the error should not be auto-retried,
	// but is eligible for manual retry.
	RetryPolicyManual RetryPolicy = retrier.RetryPolicyManual

	// RetryPolicyNever indicates a permanent failure that should not be retried.
	RetryPolicyNever RetryPolicy = retrier.RetryPolicyNever
)

type RetryableError 

type RetryableError = retrier.RetryableError

RetryableError is an interface for errors that can provide retry policy hints. Implement this interface on your error types to control retry behavior.

This is a re-export of retrier.RetryableError for convenience, so users don't need to import the retrier package separately.

type SharedBackoffOption 

type SharedBackoffOption func(*backoffConfig)

SharedBackoffOption is a functional option for shared backoff configuration.

func WithBackoffInitialDuration 

func WithBackoffInitialDuration(d time.Duration) SharedBackoffOption

WithBackoffInitialDuration sets the initial backoff duration for shared config.

func WithBackoffJitter 

func WithBackoffJitter(d time.Duration) SharedBackoffOption

WithBackoffJitter sets the jitter for shared backoff config.

func WithBackoffMaxDuration 

func WithBackoffMaxDuration(d time.Duration) SharedBackoffOption

WithBackoffMaxDuration sets the maximum backoff duration cap for shared config.

func WithBackoffMultiplier 

func WithBackoffMultiplier(m float64) SharedBackoffOption

WithBackoffMultiplier sets the exponential backoff multiplier for shared config.

type StageError 

type StageError interface {
	error
}

StageError is the primary error interface for the stage

type StageEvent 

type StageEvent struct {
	// Type indicates the event type: start, progress, complete, error.
	Type StageEventType

	// Duration of the stage (for complete events).
	Duration time.Duration

	// InputSummary is a truncated summary of input (for large inputs).
	InputSummary string

	// OutputSummary is a truncated summary of output (for large outputs).
	OutputSummary string

	// Additional structured fields.
	Fields FieldMap
}

StageEvent represents a pipeline stage event.

type StageEventType 

type StageEventType string

StageEventType represents the type of a pipeline stage event. 

const (
	// EventTypeStart indicates a stage has started.
	EventTypeStart StageEventType = "start"
	// EventTypeComplete indicates a stage has completed successfully.
	EventTypeComplete StageEventType = "complete"
	// EventTypeError indicates a stage has encountered an error.
	EventTypeError StageEventType = "error"
	// EventTypeProgress indicates a stage has reported progress.
	EventTypeProgress StageEventType = "progress"
)

type StageRecord 

type StageRecord struct {
	// Status indicates whether the stage succeeded or failed
	Status StageStatus `json:"status"`

	// Output contains the stage result (for successful stages)
	Output any `json:"output,omitempty"`

	// Input contains the input data (for retry capability)
	// For chained stages, this is the output from the previous stage
	Input any `json:"input,omitempty"`

	// Error contains the error information (for failed stages)
	// Must implement json.Marshaler for serialization
	Error any `json:"error,omitempty"`

	// Timestamp when the stage was executed
	Timestamp time.Time `json:"timestamp"`

	// Attempts counts how many times this stage has been attempted
	Attempts int `json:"attempts"`
}

StageRecord wraps stage results with metadata for journaling. This allows the journal to track both successful and failed stages, enabling retry capabilities for failed stages.

func GetStageRecord 

func GetStageRecord(runner *StageRunner, name string) (StageRecord, bool)

GetStageRecord returns the full record for a stage. This can be used to inspect the error and attempts count of a failed stage.

Example usage: 

record, ok := gopipeline.GetStageRecord(runner, "fetch-data")
if ok && record.Status == gopipeline.StatusFailed {
    fmt.Printf("Stage failed after %d attempts: %v\n", record.Attempts, record.Error)
}

type StageResult 

type StageResult[T any] struct {
	// contains filtered or unexported fields
}

func All 

func All[T any](
	ctx context.Context,
	runner *StageRunner,
	tasks ...AllTask[T],
) StageResult[[]T]

All runs multiple independent stage chains concurrently. Each task can have completely different logic, inputs, and outputs.

Unlike Pool, which processes homogeneous items with a worker pool, All runs a fixed set of different pipelines simultaneously, with one goroutine per task.

Features:

  • Error Handling: Configurable via WithErrorStrategy (FailFast stops all tasks on first error, CollectAll continues and aggregates errors)
  • Pause/Resume: Tasks respect pipeline pause state before starting
  • Progress: Completion-based progress via WithMultiTaskProgressHandler

Example usage: 

// Run 3 completely different pipelines concurrently
results := gopipeline.All(ctx, runner,
    func() gopipeline.StageResult[Result] {
        // Pipeline A: Fetch from API
        return gopipeline.Process(ctx, runner, "fetch-api", fetchFromAPI)
    },
    func() gopipeline.StageResult[Result] {
        // Pipeline B: Parse local files
        return gopipeline.Process(ctx, runner, "parse-local", parseFiles)
    },
    func() gopipeline.StageResult[Result] {
        // Pipeline C: Write to database
        return gopipeline.Process(ctx, runner, "write-db", writeToDB)
    },
)

func NewStageResult 

func NewStageResult[T any](output T, error StageError) StageResult[T]

func Pool 

func Pool[I any, O any](
	ctx context.Context,
	runner *StageRunner,
	baseName string,
	inputs []I,
	config PoolConfig,
	task PoolTask[I, O],
) StageResult[[]O]

Pool processes multiple items concurrently using a worker pool. It spawns the specified number of workers that process inputs from a shared channel.

Features:

  • Journal/Replay: Each stage uses a unique name (baseName [index+1])
  • Pause/Resume: Workers respect pipeline pause state
  • Error Handling: Configurable via WithErrorStrategy
  • Progress: Aggregated progress via WithMultiTaskProgressHandler

Example usage: 

results := gopipeline.Pool(ctx, runner, "download", urls,
    gopipeline.PoolConfig{Workers: 4},
    func(ctx context.Context, runner *gopipeline.StageRunner, idx int, url string, stageName string) gopipeline.StageResult[FileResult] {
        return gopipeline.Process(ctx, runner, stageName, func(progress gopipeline.ProgressFunc) gopipeline.StageResult[FileResult] {
            return downloadFile(url)
        })
    })

func Process 

func Process[T any](ctx context.Context, s *StageRunner, name string, task func(progress ProgressFunc) StageResult[T]) StageResult[T]

func Then 

func Then[I, O any](name string, input StageResult[I], task func(I, ProgressFunc) StageResult[O]) StageResult[O]

Then chains two pipeline stages. It takes the result of a previous stage and executes the current task only if the previous stage succeeded. If the previous stage failed, Then skips the current task and returns the error directly.

func ThenContinue 

func ThenContinue[I, O any](name string, input StageResult[I], task func(I, StageError, ProgressFunc) StageResult[O]) StageResult[O]

ThenContinue chains two pipeline stages, continuing execution even if the previous stage failed. Unlike Then, which skips execution on error, ThenContinue always executes the task. The task receives both the output (which may be zero value on error) and the error (which may be nil on success).

This is particularly useful with CollectAll error strategy, where partial results are preserved alongside errors. The task can inspect the error and decide how to handle partial results.

Example usage: 

// Archive downloaded files even if some downloads failed
archiveStage := gopipeline.ThenContinue("archive", downloadStage,
    func(results []FileResult, err gopipeline.StageError, progress gopipeline.ProgressFunc) gopipeline.StageResult[ArchiveResult] {
        if err != nil {
            // Handle partial failure - some files may have downloaded successfully
            if partial, ok := err.(*gopipeline.PartialResultsError); ok {
                fmt.Printf("Downloaded %d files, %d failed\n", len(partial.PartialResults()), len(partial.Errors()))
            }
        }
        return archiver.Archive(results)
    })

func ThenEither 

func ThenEither[I, O any](name string, input StageResult[I], condition bool, taskA func(I, ProgressFunc) StageResult[O], taskB func(I, ProgressFunc) StageResult[O]) StageResult[O]

ThenEither executes one of two stages based on a condition. This is designed for middle-of-workflow conditional stages where both branches produce the same output type but may have different implementations.

Both branches are tracked in the journal with appropriate naming:

  • taskA is named with the provided name when condition is true
  • taskB is named with the provided name when condition is false

Example usage: 

// Conditional transformation in the middle of pipeline
transformStage := gopipeline.ThenEither("transform", fetchStage, *transformEnabled,
    func(fr FetchResult, progress gopipeline.ProgressFunc) gopipeline.StageResult[TransformedResult] {
        return transformer.Transform(fr) // full transform
    },
    func(fr FetchResult, progress gopipeline.ProgressFunc) gopipeline.StageResult[TransformedResult] {
        // Passthrough alternative
        return gopipeline.NewStageResult(TransformedResult{Data: fr.Body, Skipped: true}, nil)
    },
)
// Can chain more stages after transformStage
saveStage := gopipeline.Then("save", transformStage, saveTask)

func ThenOptionalEnd 

func ThenOptionalEnd[I any](name string, input StageResult[I], enabled bool, task func(I, ProgressFunc) StageResult[any]) StageResult[any]

ThenOptionalEnd executes an optional terminal stage. When enabled=false, the stage is skipped entirely and returns an empty result. This is designed for end-of-pipeline stages where the result type doesn't matter.

Note: This returns StageResult[any] and cannot be chained to subsequent stages. For middle-of-workflow optional stages, use ThenEither instead.

Example usage: 

// Optional archiving at the end of pipeline
archiveStage := gopipeline.ThenOptionalEnd("archive", downloadStage, *archiveEnabled,
    func(results []FileResult, progress gopipeline.ProgressFunc) gopipeline.StageResult[any] {
        return archiver.Archive(results)
    })

func (StageResult[T]) Decompose 

func (r StageResult[T]) Decompose() (T, StageError)

Decompose returns the result as a tuple (output, error). This provides idiomatic Go error handling for traditionalists: 

output, err := gopipeline.Process(ctx, ... ).Decompose()
if err != nil {
    // handle error
}

func (StageResult[T]) Error 

func (r StageResult[T]) Error() StageError

func (StageResult[T]) Output 

func (r StageResult[T]) Output() T

func (StageResult[T]) Unwrap 

func (r StageResult[T]) Unwrap() T

Unwrap returns the successful value or panics if failed. Use only when failure is impossible or should crash.

func (StageResult[T]) UnwrapOr 

func (r StageResult[T]) UnwrapOr(defaultOutput T) T

UnwrapOr returns the successful value, or the provided default if failed. Perfect for fallback configurations: 

cacheTTL := gopipeline.Process(ctx, ... ).UnwrapOr(defaultTTL)

type StageRunner 

type StageRunner struct {
	// contains filtered or unexported fields
}

func NewStageRunner 

func NewStageRunner(journal Journal, opts ...StageRunnerOption) (*StageRunner, error)

func (*StageRunner) Backoff 

func (s *StageRunner) Backoff(ctx context.Context, resource string, opts ...BackoffOption)

Backoff triggers an exponential backoff for the specified resource. Use this when a remote server returns 429 Too Many Requests or 5xx errors. If rate limiting is not enabled, this is a no-op.

Example usage: 

stage.Backoff(ctx, "api.example.com")
stage.Backoff(ctx, "api.example.com", gopipeline.BackoffWithServerDelay(5*time.Second))

func (*StageRunner) IsPaused 

func (s *StageRunner) IsPaused() bool

IsPaused returns true if the pipeline is currently paused.

func (*StageRunner) Logger 

func (s *StageRunner) Logger() DebugLogger

func (*StageRunner) Pause 

func (s *StageRunner) Pause(mode PauseMode)

Pause stops pipeline execution.

  • PauseGraceful: Waits for current stage to complete before pausing. New stages will block until Resume() is called.
  • PauseImmediate: Interrupts current stage immediately via context cancellation. The running stage must check ctx.Done() to handle interruption.

Example usage: 

// Graceful pause - let current stage finish
runner.Pause(gopipeline.PauseGraceful)

// Immediate pause - interrupt current stage
runner.Pause(gopipeline.PauseImmediate)

func (*StageRunner) ResetBackoff 

func (s *StageRunner) ResetBackoff(resource string)

ResetBackoff clears the backoff state for the specified resource. Call this when the remote service has recovered. If rate limiting is not enabled, this is a no-op.

func (*StageRunner) Resume 

func (s *StageRunner) Resume()

Resume continues pipeline execution after a pause. All blocked stages will be unblocked and continue processing.

func (*StageRunner) SetBaseDelay 

func (s *StageRunner) SetBaseDelay(delay time.Duration)

SetBaseDelay sets the global base delay for rate limiting. This can be called after StageRunner creation to adjust delays dynamically. If rate limiting is not enabled, this is a no-op.

func (*StageRunner) SetResourceDelay 

func (s *StageRunner) SetResourceDelay(resource string, delay time.Duration)

SetResourceDelay sets a specific delay for a given resource. This overrides the base delay for that resource. If rate limiting is not enabled, this is a no-op.

func (*StageRunner) Wait 

func (s *StageRunner) Wait(ctx context.Context, resource string) error

Wait blocks until the rate limiter allows the operation to proceed. If rate limiting is not enabled, this is a no-op and returns nil immediately. Returns an error if the context is cancelled while waiting.

Example usage: 

if err := stage.Wait(ctx, "api.example.com"); err != nil {
    return nil, err
}

type StageRunnerOption 

type StageRunnerOption func(*stageRunnerConfig) error

StageRunnerOption is a functional option for configuring pipeline's stage runner.

func WithBackoff 

func WithBackoff(opts ...SharedBackoffOption) StageRunnerOption

WithBackoff sets shared backoff configuration used by both rate limiter and retry. This provides sensible defaults that can be overridden by feature-specific options.

Example usage: 

WithBackoff(
    WithBackoffInitialDuration(1*time.Second),
    WithBackoffMultiplier(2.0),
    WithBackoffMaxDuration(1*time.Minute),
    WithBackoffJitter(100*time.Millisecond),
)

func WithDebugLogger 

func WithDebugLogger(opts ...LoggerOption) StageRunnerOption

WithDebugLogger enables debug logging with the provided options. If no options are provided, defaults are used (FormatLogstash, stdout only). To disable debug logging entirely, simply omit this option.

Example usage: 

// Debug logger to stdout only
WithDebugLogger(WithFormat(FormatText))

// Debug logger to file AND stdout
WithDebugLogger(WithFormat(FormatJSON), WithOutputFile("logs/pipeline.log"))

// Debug logger with custom level and pre-populated fields
WithDebugLogger(
    WithFormat(FormatLogfmt),
    WithMinLevel(dlog.LevelInfo),
    WithOutputFile("logs/pipeline.log"),
    WithLoggerFields(dlog.FieldMap{"service": "scraper"}),
)

func WithErrorStrategy 

func WithErrorStrategy(strategy ErrorStrategy) StageRunnerOption

WithErrorStrategy sets how multi-task execution handles errors. - FailFast (default): Stop all tasks immediately when one fails. - CollectAll: Continue processing, collect all errors.

Example usage: 

// Fail-fast (default behavior)
WithErrorStrategy(gopipeline.FailFast)

// Collect all errors
WithErrorStrategy(gopipeline.CollectAll)

func WithMultiTaskProgressHandler 

func WithMultiTaskProgressHandler(handler MultiTaskProgressHandler) StageRunnerOption

WithMultiTaskProgressHandler sets a handler for aggregated progress across concurrent tasks. This is useful for displaying overall progress when processing multiple items concurrently.

Example usage: 

WithMultiTaskProgressHandler(func(stage string, completed, total int, percent float64, fields gopipeline.FieldMap) {
    fmt.Printf("\r%s: %d/%d complete (%.1f%%)", stage, completed, total, percent)
    if completed >= total {
        fmt.Println()
    }
})

func WithPoolFailureMode 

func WithPoolFailureMode(mode PoolFailureMode) StageRunnerOption

WithPoolFailureMode sets how pool failures are tracked in the journal. - PoolFailurePoolLevel (default): Track entire pool as one failure unit.

Example usage: 

WithPoolFailureMode(gopipeline.PoolFailurePoolLevel)

func WithProgressHandler 

func WithProgressHandler(handler ProgressHandler) StageRunnerOption

WithProgressHandler sets a custom handler for progress events. This allows custom handling of progress updates, such as displaying a CLI progress bar. The handler receives the stage name, progress percent (0-100), and any additional fields.

Example usage: 

WithProgressHandler(func(stage string, percent float64, fields gopipeline.FieldMap) {
    fmt.Printf("\r%s: %.0f%% complete", stage, percent)
})

func WithRateLimiter 

func WithRateLimiter(opts ...RateLimiterOption) StageRunnerOption

WithRateLimiter enables rate limiting with the provided options. If no options are provided, rate limiting is enabled with sensible defaults. To disable rate limiting entirely, simply omit this option.

Example usage: 

// Basic rate limiting with 2 second base delay
WithRateLimiter(WithBaseDelay(2*time.Second))

// Rate limiting with jitter and exponential backoff
WithRateLimiter(
    WithBaseDelay(500*time.Millisecond),
    WithJitter(100*time.Millisecond),
    WithMultiplier(2.0),
    WithMaxDuration(30*time.Second),
)

// Per-resource delay overrides
WithRateLimiter(
    WithBaseDelay(1*time.Second),
    WithResourceDelay("strict.api.com", 5*time.Second),
)

func WithRetry 

func WithRetry(opts ...RetryOption) StageRunnerOption

WithRetry enables auto-retry with the provided options. If no options are provided, retry is enabled with sensible defaults. By default, only errors implementing RetryableError with RetryPolicyAuto will be retried.

Example usage: 

// Basic retry with defaults (3 attempts, fail-fast)
WithRetry()

// Custom retry configuration
WithRetry(
    WithMaxAttempts(5),
    WithRetryPolicy(retrier.RetryPolicyAuto),  // retry all standard errors
)

// Retry with custom backoff (overrides shared backoff config)
WithRetry(
    WithMaxAttempts(10),
    WithRetryInitialDuration(500*time.Millisecond),
    WithRetryJitter(50*time.Millisecond),
)

type StageStatus 

type StageStatus string

StageStatus represents the status of a stage 

const (
	// StatusSuccess indicates the stage completed successfully
	StatusSuccess StageStatus = "success"
	// StatusFailed indicates the stage failed
	StatusFailed StageStatus = "failed"
)

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.