persistedretry

Documentation

Overview

Copyright (c) 2016-2019 Uber Technologies, Inc.

Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

Copyright (c) 2016-2019 Uber Technologies, Inc.

Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

Copyright (c) 2016-2019 Uber Technologies, Inc.

Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

Copyright (c) 2016-2019 Uber Technologies, Inc.

Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

Index

• Variables
• type Config
• type Executor
• type Manager
  • func NewManager(config Config, stats tally.Scope, store Store, executor Executor) (Manager, error)
• type Store
• type Task

Variables

var (
	ErrTaskExists   = errors.New("task already exists in store")
	ErrTaskNotFound = errors.New("task not found")
)

Store errors.

var ErrManagerClosed = errors.New("manager closed")

ErrManagerClosed is returned when Add is called on a closed manager.

Types

type Config

type Config struct {
	IncomingBuffer int `yaml:"incoming_buffer"`
	RetryBuffer    int `yaml:"retry_buffer"`

	NumIncomingWorkers int `yaml:"num_incoming_workers"`
	NumRetryWorkers    int `yaml:"num_retry_workers"`

	// Max rate of task execution across all workers.
	MaxTaskThroughput time.Duration `yaml:"max_task_throughput"`

	// Interval at which failed tasks should be retried.
	RetryInterval time.Duration `yaml:"retry_interval"`

	// Interval at which retries should be polled from storage.
	PollRetriesInterval time.Duration `yaml:"poll_retries_interval"`

	// Flags that zero-value channel sizes should not have defaults applied.
	Testing bool
}

Config defines Manager configuration.

type Executor

type Executor interface {
	Exec(Task) error
	Name() string
}

Executor executes tasks.

type Manager

type Manager interface {
	Add(Task) error
	SyncExec(Task) error
	Close()
	Find(query interface{}) ([]Task, error)
}

Manager defines interface for a persisted retry manager.

func NewManager

func NewManager(
	config Config, stats tally.Scope, store Store, executor Executor) (Manager, error)

NewManager creates a new Manager.

type Store

type Store interface {
	// AddPending adds a new task as pending in the store. Implementations should
	// return ErrTaskExists if the task is already in the store.
	AddPending(Task) error

	// AddFailed adds a new task as failed in the store. Implementations should
	// return ErrTaskExists if the task is already in the store.
	AddFailed(Task) error

	// MarkPending marks an existing task as pending.
	MarkPending(Task) error

	// MarkFailed marks an existing task as failed.
	MarkFailed(Task) error

	// GetPending returns all pending Tasks.
	GetPending() ([]Task, error)

	// GetFailed returns all failed Tasks.
	GetFailed() ([]Task, error)

	// Remove removes a task from the store.
	Remove(Task) error

	// Find returns tasks which match a query.
	Find(query interface{}) ([]Task, error)
}

Store provides persisted storage for tasks.

type Task

type Task interface {
	GetLastAttempt() time.Time
	GetFailures() int
	Ready() bool

	// Tags returns tags describing the context of the task, which can be
	// included on metrics to group related instances of a task.
	Tags() map[string]string
}

Task represents a single unit of work which must eventually succeed.