Second review of the night's own code: a stray clears when a later removal works (N2), the start-up check never removes this process's own copy (N3), one copy per destination at a time (N4), RunTier2 refuses during a restore for every caller (N5), the import refusal no longer advises removal (N9); red-proved, race-clean; README; REPORT; unreleased
gates / gates (push) Successful in 1m40s
gates / gates (push) Successful in 1m40s
This commit is contained in:
@@ -87,11 +87,39 @@ var tier2Docker = func(ctx context.Context, args ...string) ([]byte, error) {
|
||||
}
|
||||
|
||||
// The controller's own record of copies (package-level: one controller, one backup manager per process).
|
||||
// - inFlight: app → copies running now (capped or in-process);
|
||||
// - running: the container names THIS process runs now — the start-up check never removes one of them;
|
||||
// - strays: names whose removal failed — cleared when a later removal of that name works;
|
||||
// - startupPending: the start-up check has not succeeded yet.
|
||||
//
|
||||
// Restores are refused while any of these is non-empty (review 2026-10-23 night, N2/N3: a stray used to block every
|
||||
// restore until a restart, and the start-up retry used to remove this process's own running copies).
|
||||
var tier2Copies = struct {
|
||||
sync.Mutex
|
||||
inFlight map[string]int // app → copies running now (capped or in-process)
|
||||
stray bool // a container this process could not remove may still run, or the start-up removal has not succeeded
|
||||
}{inFlight: map[string]int{}}
|
||||
inFlight map[string]int
|
||||
running map[string]bool
|
||||
strays map[string]bool
|
||||
startupPending bool
|
||||
retrying bool
|
||||
}{inFlight: map[string]int{}, running: map[string]bool{}, strays: map[string]bool{}}
|
||||
|
||||
// tier2CopyLocks serializes copies into the SAME destination within this process (review N4): a second run waits for
|
||||
// the first instead of removing its container by name (which killed it mid-copy and mailed a false failure).
|
||||
var tier2CopyLocks = struct {
|
||||
sync.Mutex
|
||||
m map[string]*sync.Mutex
|
||||
}{m: map[string]*sync.Mutex{}}
|
||||
|
||||
func tier2CopyLock(name string) *sync.Mutex {
|
||||
tier2CopyLocks.Lock()
|
||||
defer tier2CopyLocks.Unlock()
|
||||
l := tier2CopyLocks.m[name]
|
||||
if l == nil {
|
||||
l = &sync.Mutex{}
|
||||
tier2CopyLocks.m[name] = l
|
||||
}
|
||||
return l
|
||||
}
|
||||
|
||||
func tier2CopyBegin(app string) {
|
||||
tier2Copies.Lock()
|
||||
@@ -107,55 +135,131 @@ func tier2CopyEnd(app string) {
|
||||
tier2Copies.Unlock()
|
||||
}
|
||||
|
||||
func tier2SetStray(v bool) {
|
||||
func tier2MarkRunning(name string, on bool) {
|
||||
tier2Copies.Lock()
|
||||
tier2Copies.stray = v
|
||||
if on {
|
||||
tier2Copies.running[name] = true
|
||||
} else {
|
||||
delete(tier2Copies.running, name)
|
||||
}
|
||||
tier2Copies.Unlock()
|
||||
}
|
||||
|
||||
// tier2RemoveCopy removes the copy container `name` (no error when it does not exist). It reports whether the
|
||||
// container is surely gone; when not, the stray flag is set, so restores stay refused until a removal succeeds.
|
||||
// container is surely gone; when not, the name is recorded as a stray (restores refused) and retried every minute.
|
||||
func tier2RemoveCopy(name, why string) bool {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
|
||||
defer cancel()
|
||||
out, err := tier2Docker(ctx, "rm", "-f", name)
|
||||
if err != nil && !strings.Contains(strings.ToLower(string(out)), "no such container") {
|
||||
tier2Logf("[ERROR] [tier2] could not remove the copy container %s (%s): %v %s — restores stay refused until it is removed",
|
||||
tier2Logf("[ERROR] [tier2] could not remove the copy container %s (%s): %v %s — restores stay refused until it is removed (retried every minute)",
|
||||
name, why, err, strings.TrimSpace(string(out)))
|
||||
tier2SetStray(true)
|
||||
tier2Copies.Lock()
|
||||
tier2Copies.strays[name] = true
|
||||
tier2Copies.Unlock()
|
||||
tier2StartRetry(context.Background())
|
||||
return false
|
||||
}
|
||||
tier2Copies.Lock()
|
||||
delete(tier2Copies.strays, name)
|
||||
tier2Copies.Unlock()
|
||||
return true
|
||||
}
|
||||
|
||||
// ClearLeftoverTier2Copies removes every labelled copy container — a copy a previous controller process started and
|
||||
// lost (restart, self-update, crash). It runs at controller start, BEFORE the scheduler and the web server; each
|
||||
// removal is logged. Until it succeeds, restores are refused; on failure it retries every minute in the background
|
||||
// until it succeeds or ctx ends.
|
||||
func ClearLeftoverTier2Copies(ctx context.Context) error {
|
||||
tier2SetStray(true)
|
||||
err := clearLeftoverTier2CopiesOnce(ctx)
|
||||
if err == nil {
|
||||
tier2SetStray(false)
|
||||
return nil
|
||||
// tier2RetryEvery is the retry period for strays and a failed start-up check; tier2RetryOff (tests only) keeps the
|
||||
// background retry from starting at all — a timer that outlives a test would reach the real Docker of the host.
|
||||
var (
|
||||
tier2RetryEvery = time.Minute
|
||||
tier2RetryOff = false
|
||||
)
|
||||
|
||||
// tier2StartRetry starts ONE background retry (if none runs): every tier2RetryEvery it redoes a failed start-up check
|
||||
// and the removal of every stray, and ends when nothing is left or ctx ends.
|
||||
func tier2StartRetry(ctx context.Context) {
|
||||
tier2Copies.Lock()
|
||||
if tier2Copies.retrying || tier2RetryOff {
|
||||
tier2Copies.Unlock()
|
||||
return
|
||||
}
|
||||
tier2Logf("[ERROR] [tier2] the start-up check for a leftover second-drive copy failed: %v — restores are refused until it succeeds; retrying every minute", err)
|
||||
tier2Copies.retrying = true
|
||||
tier2Copies.Unlock()
|
||||
every := tier2RetryEvery
|
||||
go func() {
|
||||
t := time.NewTicker(time.Minute)
|
||||
t := time.NewTicker(every)
|
||||
defer t.Stop()
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
tier2Copies.Lock()
|
||||
tier2Copies.retrying = false
|
||||
tier2Copies.Unlock()
|
||||
return
|
||||
case <-t.C:
|
||||
if clearLeftoverTier2CopiesOnce(ctx) == nil {
|
||||
tier2SetStray(false)
|
||||
tier2Logf("[INFO] [tier2] the leftover second-drive copy check succeeded — restores allowed again")
|
||||
return
|
||||
}
|
||||
}
|
||||
if tier2RetryOnce(ctx) {
|
||||
tier2Logf("[INFO] [tier2] no leftover or stray second-drive copy container is left — restores allowed again")
|
||||
return
|
||||
}
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
// tier2RetryOnce does one retry round; it reports (and clears `retrying`) when nothing is left to retry.
|
||||
func tier2RetryOnce(ctx context.Context) bool {
|
||||
tier2Copies.Lock()
|
||||
pending := tier2Copies.startupPending
|
||||
var names []string
|
||||
for n := range tier2Copies.strays {
|
||||
names = append(names, n)
|
||||
}
|
||||
tier2Copies.Unlock()
|
||||
if pending && clearLeftoverTier2CopiesOnce(ctx) == nil {
|
||||
tier2Copies.Lock()
|
||||
tier2Copies.startupPending = false
|
||||
tier2Copies.Unlock()
|
||||
}
|
||||
for _, n := range names {
|
||||
tier2RemoveCopyQuiet(ctx, n)
|
||||
}
|
||||
tier2Copies.Lock()
|
||||
defer tier2Copies.Unlock()
|
||||
if !tier2Copies.startupPending && len(tier2Copies.strays) == 0 {
|
||||
tier2Copies.retrying = false
|
||||
return true
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// tier2RemoveCopyQuiet is the retry's removal: a failure keeps the stray without starting another retry.
|
||||
func tier2RemoveCopyQuiet(ctx context.Context, name string) {
|
||||
rctx, cancel := context.WithTimeout(ctx, 60*time.Second)
|
||||
defer cancel()
|
||||
out, err := tier2Docker(rctx, "rm", "-f", name)
|
||||
if err != nil && !strings.Contains(strings.ToLower(string(out)), "no such container") {
|
||||
return
|
||||
}
|
||||
tier2Copies.Lock()
|
||||
delete(tier2Copies.strays, name)
|
||||
tier2Copies.Unlock()
|
||||
tier2Logf("[INFO] [tier2] the stray copy container %s is removed", name)
|
||||
}
|
||||
|
||||
// ClearLeftoverTier2Copies removes every labelled copy container a previous controller process started and lost
|
||||
// (restart, self-update, crash). It runs at controller start, BEFORE the scheduler and the web server; each removal is
|
||||
// logged. Until it succeeds, restores are refused; on failure it is retried every minute in the background.
|
||||
func ClearLeftoverTier2Copies(ctx context.Context) error {
|
||||
tier2Copies.Lock()
|
||||
tier2Copies.startupPending = true
|
||||
tier2Copies.Unlock()
|
||||
err := clearLeftoverTier2CopiesOnce(ctx)
|
||||
if err == nil {
|
||||
tier2Copies.Lock()
|
||||
tier2Copies.startupPending = false
|
||||
tier2Copies.Unlock()
|
||||
return nil
|
||||
}
|
||||
tier2Logf("[ERROR] [tier2] the start-up check for a leftover second-drive copy failed: %v — restores are refused until it succeeds; retrying every minute", err)
|
||||
tier2StartRetry(ctx)
|
||||
return err
|
||||
}
|
||||
|
||||
@@ -166,7 +270,14 @@ func clearLeftoverTier2CopiesOnce(ctx context.Context) error {
|
||||
if err != nil {
|
||||
return fmt.Errorf("docker ps: %v %s", err, strings.TrimSpace(string(out)))
|
||||
}
|
||||
names := strings.Fields(string(out))
|
||||
var names []string
|
||||
tier2Copies.Lock()
|
||||
for _, n := range strings.Fields(string(out)) {
|
||||
if !tier2Copies.running[n] { // never a copy THIS process runs (review N3)
|
||||
names = append(names, n)
|
||||
}
|
||||
}
|
||||
tier2Copies.Unlock()
|
||||
if len(names) == 0 {
|
||||
tier2Logf("[INFO] [tier2] start-up: no leftover second-drive copy container")
|
||||
return nil
|
||||
@@ -195,7 +306,7 @@ func clearLeftoverTier2CopiesOnce(ctx context.Context) error {
|
||||
func (m *Manager) Tier2CopyBlocksRestore() (app string, blocked bool) {
|
||||
tier2Copies.Lock()
|
||||
defer tier2Copies.Unlock()
|
||||
if tier2Copies.stray {
|
||||
if tier2Copies.startupPending || len(tier2Copies.strays) > 0 {
|
||||
return "", true
|
||||
}
|
||||
for a := range tier2Copies.inFlight {
|
||||
@@ -204,6 +315,14 @@ func (m *Manager) Tier2CopyBlocksRestore() (app string, blocked bool) {
|
||||
return "", false
|
||||
}
|
||||
|
||||
// tier2ResetForTest empties the record (tests only).
|
||||
func tier2ResetForTest() {
|
||||
tier2Copies.Lock()
|
||||
tier2Copies.inFlight, tier2Copies.running, tier2Copies.strays = map[string]int{}, map[string]bool{}, map[string]bool{}
|
||||
tier2Copies.startupPending, tier2Copies.retrying = false, false
|
||||
tier2Copies.Unlock()
|
||||
}
|
||||
|
||||
// MarkTier2CopyForTest records a copy of app as running until the returned func is called — a test-only seam (another
|
||||
// package's test exercises the restore refusal without a real copy), in the spirit of AcquireRunningForTest.
|
||||
func MarkTier2CopyForTest(app string) func() {
|
||||
|
||||
Reference in New Issue
Block a user