Multi-store failover

store/failover provides four stateless functions over a list of drivers. The order of the driver slice is the priority: which drivers take part, which are healthy, and which ones hold a copy of an object are all decided by the caller, who passes them in already sorted. The package itself keeps no state.

func Write(ctx context.Context, ds []driver.Driver, key string, r io.ReaderAt, size int64,
	opts WriteOptions) ([]Attempt, error)
func Open(ctx context.Context, ds []driver.Driver, key string) (io.ReadCloser, []Attempt, error)
func Probe(ctx context.Context, d driver.Driver) error
func DeepProbe(ctx context.Context, d driver.Driver) error

Write and Open both return []Attempt, recording every driver actually tried, in the order tried:

Write: write in order until N copies succeedCopy Anchor Link

attempts, err := failover.Write(ctx, ds, key, bytes.NewReader(data), int64(len(data)),
	failover.WriteOptions{Need: 2, AttemptTimeout: time.Minute})

• Drivers are written one at a time in slice order, never concurrently. When a driver fails, the next one is tried; once Need copies succeed the call returns and later drivers are not touched.
• Each attempt reads the data from the beginning. The source is an io.ReaderAt plus size, and every attempt reads [0, size) independently, so a driver that fails halfway does not affect the next one. *bytes.Reader and *os.File both satisfy io.ReaderAt.
• Cancelling ctx stops immediately: ctx is checked before each attempt; once it is cancelled no further drivers are tried and ctx.Err() is returned, not ErrInsufficient. A failure caused by cancellation says nothing about the drivers.
• Attempts with Err == nil are the drivers that now hold a copy. Even on success, attempts may include failed attempts.
• Failed writes are not cleaned up: a failed driver may be left with a partial object. Delete it yourself if needed, or let reconciliation deal with it.
• A Need larger than the number of drivers is not rejected up front; all drivers are tried, then ErrInsufficient is returned.

Open: read from the first readable driverCopy Anchor Link

rc, attempts, err := failover.Open(ctx, ds, key)
if err != nil {
	return err
}
defer func() { _ = rc.Close() }()

• Drivers are opened in slice order and the first successful open is returned immediately; later drivers are not touched. The caller closes the reader.
• On success, the last element of attempts is the driver serving the reader; the earlier ones are failed attempts, useful for logs and metrics.
• Fallback only happens while opening. If the reader fails partway through reading, it does not switch to another driver.
• There is no per-attempt timeout, only ctx; ctx is checked before each attempt, and once it is cancelled ctx.Err() is returned. With rclone drivers, ctx usually also governs the returned reader (for HTTP-based backends, cancelling ctx aborts the response body), so do not cancel it before you finish reading.

When every driver fails, there are two kinds of error:

An empty driver slice returns ErrNoDrivers.

The two kinds are mutually exclusive: errors.Is(err, driver.ErrNotFound) is true only when every driver confirmed the object does not exist, and always false when some driver is broken. An ErrUnavailable error does not wrap the drivers' original errors; to examine why a particular driver failed, look at the returned attempts:

rc, attempts, err := failover.Open(ctx, ds, key)
switch {
case errors.Is(err, driver.ErrNotFound):
	// Every driver confirmed the object does not exist: e.g. HTTP 404
case errors.Is(err, failover.ErrUnavailable):
	// Some driver is broken and the object may still exist: treat as temporarily unavailable, e.g. HTTP 503
	for _, a := range attempts {
		log.Printf("driver %d: %v", a.Index, a.Err)
	}
case err != nil:
	// ErrNoDrivers, or ctx was cancelled
default:
	defer func() { _ = rc.Close() }()
	// read from rc
}

Probe and DeepProbe: health checksCopy Anchor Link

Probe's timeout is entirely up to ctx, so always call it with a deadline. ProbeKey is .imgyoung-health-probe; by convention it is never written, so do not store data under that key.

Details of DeepProbe:

• The temporary object key is .imgyoung-probe- followed by 16 random hex characters, and the content is a fixed 19 bytes; random keys keep concurrent deep checks from interfering with each other.
• The write and the read-back are bound by ctx. The delete uses its own 10-second timeout detached from ctx cancellation, so it still tries to clean up after a failed read-back or an expired ctx.
• A failure in the write, the read-back, or the content comparison returns the corresponding error. If the read-back already failed, the delete result no longer affects the return value; if the read-back succeeded but the delete failed (with anything other than driver.ErrNotFound), the delete error is returned.
• A failed delete leaves a temporary object behind. failover.IsDeepProbeKey(key) reports whether a key is a deep check temporary object, so your cleanup logic can claim and reclaim it.

A successful read-only check does not mean the store accepts writes. A safe combination, and the one the application uses, is "downgrade only": periodic light checks can only mark a healthy store unhealthy; an unhealthy store whose light check passes must also pass a deep check (write, read, and delete all succeed) before it is marked healthy again. That way a store with read-only credentials, which passes the light check but rejects writes, is not mistaken for recovered on the next light check.

Complete exampleCopy Anchor Link

This is store/failover/example_test.go from the repository. Two local directories act as the primary and backup stores. The primary's root path sits under a regular file, so every write to it fails, simulating a broken primary; the write falls through to the backup, and the read switches to the backup as well. It runs with go test ./store/failover and prints exactly what the Output comment at the end lists (the comments in the source are in Chinese):

package failover_test

import (
	"bytes"
	"context"
	"fmt"
	"io"
	"os"
	"path/filepath"

	"github.com/x-haose/imgyoung/store/driver"
	"github.com/x-haose/imgyoung/store/failover"
	"github.com/x-haose/imgyoung/store/rclone"
)

// Example 两个本地目录充当主、备存储,主存储故障时写入顺延到备存储,读取同样切到备存储。
func Example() {
	ctx := context.Background()
	dir, err := os.MkdirTemp("", "failover-example")
	if err != nil {
		fmt.Println("mkdir:", err)
		return
	}
	defer func() { _ = os.RemoveAll(dir) }()

	// 主存储的根落在一个普通文件之下,任何写入都会失败,以此模拟主存储故障。
	blocker := filepath.Join(dir, "blocker")
	if err := os.WriteFile(blocker, nil, 0o600); err != nil {
		fmt.Println("write blocker:", err)
		return
	}
	primary, err := rclone.Open(ctx, rclone.Options{Backend: "local", Path: filepath.Join(blocker, "data")})
	if err != nil {
		fmt.Println("open primary:", err)
		return
	}
	defer func() { _ = primary.Close(ctx) }()
	backup, err := rclone.Open(ctx, rclone.Options{Backend: "local", Path: filepath.Join(dir, "backup")})
	if err != nil {
		fmt.Println("open backup:", err)
		return
	}
	defer func() { _ = backup.Close(ctx) }()
	ds := []driver.Driver{primary, backup} // 顺序即优先级

	data := []byte("hello")
	written, err := failover.Write(ctx, ds, "a.txt", bytes.NewReader(data), int64(len(data)),
		failover.WriteOptions{Need: 1})
	if err != nil {
		fmt.Println("write:", err)
		return
	}
	for _, a := range written {
		fmt.Println("write to", a.Index, "ok:", a.Err == nil)
	}

	rc, read, err := failover.Open(ctx, ds, "a.txt")
	if err != nil {
		fmt.Println("read:", err)
		return
	}
	got, err := io.ReadAll(rc)
	_ = rc.Close()
	if err != nil {
		fmt.Println("read all:", err)
		return
	}
	fmt.Println(string(got), "from", read[len(read)-1].Index)
	// Output:
	// write to 0 ok: false
	// write to 1 ok: true
	// hello from 1
}

Building your own replicationCopy Anchor Link

A common pattern is "return once the primary write succeeds and fill in the other stores in the background": call Write with Need: 1, remember the drivers that were not written, and copy to them later.

// writePrimary returns once one copy is written, reporting which drivers hold it and which still need it.
func writePrimary(ctx context.Context, ds []driver.Driver, key string, data []byte) (holders, pending []int, err error) {
	attempts, err := failover.Write(ctx, ds, key, bytes.NewReader(data), int64(len(data)),
		failover.WriteOptions{Need: 1, AttemptTimeout: time.Minute})
	if err != nil {
		return nil, nil, err
	}
	done := map[int]bool{}
	for _, a := range attempts {
		if a.Err == nil {
			done[a.Index] = true
			holders = append(holders, a.Index)
		}
	}
	for i := range ds {
		if !done[i] {
			pending = append(pending, i)
		}
	}
	return holders, pending, nil
}

Replicating one copy means reading it from a driver that holds it and writing it to the target driver, and the read step can use Open's fallback directly:

// Replicate copies key from holders (drivers that hold a copy, in priority order) to target.
// size is the object size in bytes, normally taken from your own metadata; without it, Stat a source driver first.
func Replicate(ctx context.Context, holders []driver.Driver, target driver.Driver, key string, size int64) error {
	rc, attempts, err := failover.Open(ctx, holders, key)
	if err != nil {
		return fmt.Errorf("no readable source copy: %w", err)
	}
	for _, a := range attempts[:len(attempts)-1] {
		log.Printf("source driver %d unreadable, moved on: %v", a.Index, a.Err)
	}
	err = target.Put(ctx, key, rc, size)
	if cerr := rc.Close(); err == nil {
		err = cerr
	}
	return err
}

These two functions only perform the work. To turn them into reliable replication you also need:

• Persisted replica state: store holders and pending in your own database so unfinished replication can resume after a restart.
• Retries with backoff: retry a failed Replicate later, mark it failed and alert after repeated failures, and try again once the store recovers.
• Reconciliation: periodically List every store and compare with your records in both directions. Objects in the store but not in your records are orphans (produced when the write succeeded but recording it failed); delete them after a grace period. Copies your records claim but the store lacks need to be replicated again. Only reclaim objects you can tell you wrote yourself (by your own key format, for example), and leave files other programs put in the same store alone; claim deep check leftovers with IsDeepProbeKey and delete them once past the grace period.
• Concurrency limits: replication shares the stores' bandwidth and request quotas with regular traffic, so cap how many copies run at once.

The application records replica state in the blob_channels table and runs replication on the River task queue; see Architecture.

How the application uses these primitivesCopy Anchor Link

The application's channels are a set of rclone drivers, and upload, image access, and health checks all call this package directly:

Replication jobs try the source channels one by one and write to the target channel, the same idea as Replicate above. The daily reconciliation skips ProbeKey and the system directory _imgyoung/, and only reclaims orphans whose keys have the application's object key format (criteria in Architecture); leftovers from deep checks are claimed with IsDeepProbeKey and deleted once past the 48-hour grace period, regardless of when they were written. For runtime settings, see Configuration.