irrecoverable

Documentation

Overview

Example

package main

import (
	"context"
	"errors"
	"fmt"
	"sync"
	"time"

	"github.com/koko1123/flow-go-1/module/component"
	"github.com/koko1123/flow-go-1/module/irrecoverable"
)

var ErrTriggerRestart = errors.New("restart me")
var ErrNoRestart = errors.New("fatal, no restarts")

func main() {
	// a context is mandatory in order to call RunComponent
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	// component.ComponentFactory encapsulates all of the component building logic
	// required before running Start()
	starts := 0
	componentFactory := func() (component.Component, error) {
		starts++
		return NewExampleComponent(starts), nil
	}

	// this is the place to inspect the encountered error and implement the appropriate error
	// handling behaviors, e.g. restarting the component, firing an alert to pagerduty, etc ...
	// the shutdown of the component is handled for you by RunComponent, but you may consider
	// performing additional cleanup here
	onError := func(err error) component.ErrorHandlingResult {
		// check the error type to decide whether to restart or shutdown
		if errors.Is(err, ErrTriggerRestart) {
			fmt.Printf("Restarting component after fatal error: %v\n", err)
			return component.ErrorHandlingRestart
		} else {
			fmt.Printf("An irrecoverable error occurred: %v\n", err)
			// shutdown other components. it might also make sense to just panic here
			// depending on the circumstances
			return component.ErrorHandlingStop
		}
	}

	// run the component. this is a blocking call, and will return with an error if the
	// first startup or any subsequent restart attempts fails or the context is canceled
	err := component.RunComponent(ctx, componentFactory, onError)
	if err != nil {
		fmt.Printf("Error returned from RunComponent: %v\n", err)
	}

}

// ExampleComponent is an example of a typical component
type ExampleComponent struct {
	id      int
	started chan struct{}
	ready   sync.WaitGroup
	done    sync.WaitGroup
}

func NewExampleComponent(id int) *ExampleComponent {
	return &ExampleComponent{
		id:      id,
		started: make(chan struct{}),
	}
}

// start the component and register its shutdown handler
// this component will throw an error after 20ms to demonstrate the error handling
func (c *ExampleComponent) Start(ctx irrecoverable.SignalerContext) {
	c.printMsg("Starting up")

	// do some setup...

	c.ready.Add(2)
	c.done.Add(2)

	go func() {
		c.ready.Done()
		defer c.done.Done()

		<-ctx.Done()

		c.printMsg("Shutting down")
		// do some cleanup...
	}()

	go func() {
		c.ready.Done()
		defer c.done.Done()

		select {
		case <-time.After(20 * time.Millisecond):
			// encounter irrecoverable error
			if c.id > 1 {
				ctx.Throw(ErrNoRestart)
			} else {
				ctx.Throw(ErrTriggerRestart)
			}
		case <-ctx.Done():
			c.printMsg("Cancelled by parent")
		}
	}()

	close(c.started)
}

// simply return the Started channel
// all startup processing is done in Start()
func (c *ExampleComponent) Ready() <-chan struct{} {
	ready := make(chan struct{})
	go func() {
		<-c.started
		c.ready.Wait()
		close(ready)
	}()
	return ready
}

// simply return the Stopped channel
// all shutdown processing is done in shutdownOnCancel()
func (c *ExampleComponent) Done() <-chan struct{} {
	done := make(chan struct{})
	go func() {
		<-c.started
		c.done.Wait()
		close(done)
	}()
	return done
}

func (c *ExampleComponent) printMsg(msg string) {
	fmt.Printf("[Component %d] %s\n", c.id, msg)
}

Output:

[Component 1] Starting up
[Component 1] Shutting down
Restarting component after fatal error: restart me
[Component 2] Starting up
[Component 2] Shutting down
An irrecoverable error occurred: fatal, no restarts
Error returned from RunComponent: fatal, no restarts

Index

• func Throw(ctx context.Context, err error)
• type MockSignalerContext
  • func NewMockSignalerContext(t *testing.T, ctx context.Context) *MockSignalerContext
  • func (m MockSignalerContext) Throw(err error)
• type Signaler
  • func NewSignaler() (*Signaler, <-chan error)
  • func (s *Signaler) Throw(err error)
• type SignalerContext
  • func WithSignaler(parent context.Context) (SignalerContext, <-chan error)

Examples

• Package

Functions

func Throw

func Throw(ctx context.Context, err error)

If we have an SignalerContext, we can directly ctx.Throw.

But a lot of library methods expect context.Context, & we want to pass the same w/o boilerplate Moreover, we could have built with: context.WithCancel(irrecoverable.WithSignaler(ctx, sig)), "downcasting" to context.Context. Yet, we can still type-assert and recover.

Throw can be a drop-in replacement anywhere we have a context.Context likely to support Irrecoverables. Note: this is not a method

Types

type MockSignalerContext

type MockSignalerContext struct {
	context.Context
	// contains filtered or unexported fields
}

func NewMockSignalerContext

func NewMockSignalerContext(t *testing.T, ctx context.Context) *MockSignalerContext

func (MockSignalerContext) Throw

func (m MockSignalerContext) Throw(err error)

type Signaler

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

Signaler sends the error out.

func NewSignaler

func NewSignaler() (*Signaler, <-chan error)

func (*Signaler) Throw

func (s *Signaler) Throw(err error)

Throw is a narrow drop-in replacement for panic, log.Fatal, log.Panic, etc anywhere there's something connected to the error channel. It only sends the first error it is called with to the error channel, and logs subsequent errors as unhandled.

type SignalerContext

type SignalerContext interface {
	context.Context
	Throw(err error) // delegates to the signaler
	// contains filtered or unexported methods
}

We define a constrained interface to provide a drop-in replacement for context.Context including in interfaces that compose it.

func WithSignaler

func WithSignaler(parent context.Context) (SignalerContext, <-chan error)

the One True Way of getting a SignalerContext