fix(schedule): prevent pinned startup lock collisions

internal/engine/backup_postgres_ops.go

@@ -496,6 +496,25 @@ func (e *Engine) hasFlock(ctx context.Context) bool {
 	return e.flockPresent
 }
 
+// hasScheduleFlock is deliberately stricter than hasFlock. Backup locking only
+// needs the historical short options, while generated schedule units invoke
+// /usr/bin/flock directly and need this complete util-linux long-option
+// interface. Reject an incompatible host before installing units rather than
+// discovering it when a timer fires.
+func (e *Engine) hasScheduleFlock(ctx context.Context) bool {
+	if e.scheduleFlockProbed {
+		return e.scheduleFlockPresent
+	}
+	res, err := e.T.Run(ctx, scheduleFlockProbe("/usr/bin/flock"))
+	e.scheduleFlockProbed = true
+	e.scheduleFlockPresent = err == nil && strings.TrimSpace(res.Stdout) == "ok"
+	return e.scheduleFlockPresent
+}
+
+func scheduleFlockProbe(path string) string {
+	return "command -v flock >/dev/null 2>&1 || exit; test -x " + q(path) + " || exit; help=$(" + q(path) + " --help 2>&1) || exit; for option in --conflict-exit-code --exclusive --nonblock --shared --timeout --unlock; do printf '%s\\n' \"$help\" | grep -q -- \"$option\" || exit; done; echo ok"
+}
+
 // walgLockPrefix is the flock every repository operation runs behind, as a
 // command prefix so callers that build their own docker exec can use it too.
 // Empty when the host has no flock — see hasFlock for why that is not a silent

internal/engine/bootstrap_test.go

@@ -40,7 +40,7 @@ func TestBootstrapSequence(t *testing.T) {
 	seq := strings.Join(f.Commands, "\n")
 	ordered := []string{
 		"mkdir -p",                                           // dirs
-		"> '/var/lib/ob/sample/lock'",                        // application lock
+		`link "$tmp" '/var/lib/ob/sample/lock'`,              // application lock
 		"> '/var/lib/ob/sample/fence'",                       // mutation fence
 		`"phase":"bootstrap","event":"start"`,                // durable journal boundary
 		"apt-get install -y something-host-specific",         // bootstrap hook
@@ -212,7 +212,7 @@ func TestBootstrapRefusesMissingRuntimeWithoutImplicitInstaller(t *testing.T) {
 	if runtimeCheck < 0 {
 		t.Fatalf("bootstrap did not check the runtime:\n%s", seq)
 	}
-	for _, before := range []string{"> '/var/lib/ob/sample/lock'", "> '/var/lib/ob/sample/fence'", `"phase":"bootstrap","event":"start"`} {
+	for _, before := range []string{`link "$tmp" '/var/lib/ob/sample/lock'`, "> '/var/lib/ob/sample/fence'", `"phase":"bootstrap","event":"start"`} {
 		if index := strings.Index(seq, before); index < 0 || index > runtimeCheck {
 			t.Fatalf("%q did not precede the runtime check:\n%s", before, seq)
 		}

internal/engine/deploy_test.go

@@ -604,7 +604,7 @@ func TestDeployKeepsAndExplainsTheLockWhenItRefuses(t *testing.T) {
 		t.Fatal("expected a refusal")
 	}
 	for _, c := range f.Commands {
-		if strings.Contains(c, "rm -f") && strings.Contains(c, "/lock") {
+		if strings.Contains(c, "rm -f '/var/lib/ob/sample/lock'") {
 			t.Fatalf("the lock was released over a live container:\n%s", c)
 		}
 	}

internal/engine/engine.go

@@ -88,8 +88,10 @@ type Engine struct {
 	Spec *app.Resolved
 	// flockProbed/flockPresent cache whether the target has flock, which every
 	// wal-g invocation needs to know and which cannot change mid-operation.
-	flockProbed  bool
-	flockPresent bool
+	flockProbed          bool
+	flockPresent         bool
+	scheduleFlockProbed  bool
+	scheduleFlockPresent bool
 	// triggerUnitProbed/triggerUnitPresent cache whether the host's systemd
 	// tells a timer activation from a manual one (TRIGGER_UNIT, systemd 252).
 	triggerUnitProbed  bool

internal/engine/epoch_test.go

@@ -228,7 +228,8 @@ func assertEpochAcquisition(t *testing.T, fake *transport.Fake, got int, err err
 		if err == nil {
 			t.Fatal("invalid epoch was accepted")
 		}
-		if strings.Contains(strings.Join(fake.Commands, "\n"), "set -C") {
+		commands := strings.Join(fake.Commands, "\n")
+		if strings.Contains(commands, "set -C") || strings.Contains(commands, "lock.candidate.XXXXXX") {
 			t.Fatalf("lock was created after epoch validation failed:\n%s", strings.Join(fake.Commands, "\n"))
 		}
 		return

internal/engine/job_test.go

@@ -85,7 +85,7 @@ func TestRunJobRejectsDeclarationDriftBeforeLock(t *testing.T) {
 			if err == nil || !strings.Contains(err.Error(), test.want) {
 				t.Fatalf("error = %v, want %q", err, test.want)
 			}
-			if strings.Contains(strings.Join(target.Commands, "\n"), "set -C; echo") {
+			if strings.Contains(strings.Join(target.Commands, "\n"), "lock.candidate.XXXXXX") {
 				t.Fatalf("declaration refusal acquired the app lock: %#v", target.Commands)
 			}
 		})
@@ -124,7 +124,7 @@ func TestRunJobRechecksReleaseAndRuntimeUnderLock(t *testing.T) {
 				t.Fatalf("error = %v, want %q", err, test.want)
 			}
 			commands := strings.Join(target.Commands, "\n")
-			if !strings.Contains(commands, "set -C; echo") || !strings.Contains(commands, "/fence") {
+			if !strings.Contains(commands, "lock.candidate.XXXXXX") || !strings.Contains(commands, "/fence") {
 				t.Fatalf("post-lock recheck was not lock/fence protected:\n%s", commands)
 			}
 			if strings.Contains(commands, "ONEBOX_RESULT_FILE") {
@@ -229,7 +229,7 @@ func TestRunJobKeepsTheLockWhenItRefuses(t *testing.T) {
 		t.Fatal("expected a refusal")
 	}
 	for _, c := range target.Commands {
-		if strings.Contains(c, "rm -f") && strings.Contains(c, "/lock") {
+		if strings.Contains(c, "rm -f '/var/lib/ob/sample/lock'") {
 			t.Fatalf("the lock was released over a live container:\n%s", c)
 		}
 	}
@@ -286,7 +286,7 @@ func TestRunJobKeepsTheLockWhenItCannotAskTheHost(t *testing.T) {
 		t.Fatal("an unanswerable host must refuse")
 	}
 	for _, c := range target.Commands {
-		if strings.Contains(c, "rm -f") && strings.Contains(c, "/lock") {
+		if strings.Contains(c, "rm -f '/var/lib/ob/sample/lock'") {
 			t.Fatalf("the lock was released without an answer:\n%s", c)
 		}
 	}

internal/engine/lock.go

@@ -36,6 +36,19 @@ type pinnedScheduleLeasePolicy struct {
 	conflict string
 }
 
+// scheduleRendezvousWaitSeconds lets a reader or writer already inside the
+// short schedule/deploy handoff finish without turning ordinary concurrency
+// into a missed firing or a refused operation. It does not wait for the
+// durable application lock: that lock may cover a whole deploy or exec.
+const scheduleRendezvousWaitSeconds = 10
+
+// Commands run under flock normalize expected collisions so their exit status
+// does not depend on shell or flock defaults. util-linux reserves 64–78 for its
+// own errors; keep both sentinels above that range and distinct so
+// infrastructure failures remain visible.
+const applicationLockHeldExitCode = 79
+const flockConflictExitCode = 200
+
 func (e *Engine) base() string      { return release.PathsFor(e.names()).Base }
 func (e *Engine) lockPath() string  { return e.base() + "/lock" }
 func (e *Engine) epochPath() string { return e.base() + "/epoch" }
@@ -76,15 +89,22 @@ func (e *Engine) acquireLock(ctx context.Context, deployID string, force bool, l
 			TTLSeconds: int(e.lockTTL().Seconds()), AcquiredAt: time.Now().UTC().Format(time.RFC3339),
 		}
 		b, _ := json.Marshal(meta)
-		// noclobber: the remote shell refuses the redirect if the lock exists
-		create := "set -C; echo " + q(string(b)) + " > " + q(e.lockPath()) + " 2>/dev/null"
+		create := atomicApplicationLockCreateCmd(e.lockPath(), string(b))
 		jobs, scheduleErr := e.Spec.ScheduledJobs()
 		if scheduleErr != nil {
 			return 0, scheduleErr
 		}
-		useScheduleLock := e.hasFlock(ctx)
+		useScheduleLock := e.hasScheduleFlock(ctx)
+		useLegacyScheduleLock := false
+		if !useScheduleLock {
+			// The current spec may have just removed its last schedule while an
+			// old unit is already starting. Preserve the pre-upgrade rendezvous
+			// with the short-option interface in that transition. Its ambiguous
+			// nonzero exits fail visibly below instead of being called contention.
+			useLegacyScheduleLock = e.hasFlock(ctx)
+		}
 		if len(jobs) > 0 && !useScheduleLock {
-			return 0, errors.New("scheduled jobs require flock on the target so they cannot overlap deployments; install util-linux and deploy again")
+			return 0, errors.New("scheduled jobs require a compatible util-linux flock at /usr/bin/flock so lock contention can be distinguished from host failures; install util-linux or upgrade it and deploy again")
 		}
 		if useScheduleLock {
 			// An exclusive scheduled job holds this kernel lock for its whole run;
@@ -93,7 +113,10 @@ func (e *Engine) acquireLock(ctx context.Context, deployID string, force bool, l
 			// pass its check before the other publishes ownership. Keep doing this
 			// after the last schedule is removed: an old unit may already be
 			// starting while that removal deploy begins.
-			create = "/usr/bin/flock --exclusive --nonblock --conflict-exit-code 76 " +
+			create = "/usr/bin/flock --exclusive --timeout " + strconv.Itoa(scheduleRendezvousWaitSeconds) + " --conflict-exit-code " + strconv.Itoa(flockConflictExitCode) + " " +
+				q(e.names().ScheduleRunLock()) + " /bin/sh -c " + q(create)
+		} else if useLegacyScheduleLock {
+			create = "/usr/bin/flock -x -w " + strconv.Itoa(scheduleRendezvousWaitSeconds) + " " +
 				q(e.names().ScheduleRunLock()) + " /bin/sh -c " + q(create)
 		}
 
@@ -130,8 +153,15 @@ func (e *Engine) acquireLock(ctx context.Context, deployID string, force bool, l
 			}
 			return epoch, nil
 		}
-		if res.ExitCode == 76 {
-			return 0, fmt.Errorf("deploy lock held by a scheduled job — wait for the job to finish")
+		if useScheduleLock && res.ExitCode == flockConflictExitCode {
+			return 0, fmt.Errorf("application scheduling rendezvous remained busy — wait for the current scheduled job or application operation to finish")
+		}
+		if res.ExitCode != applicationLockHeldExitCode {
+			detail := strings.TrimSpace(res.Stderr)
+			if detail == "" {
+				detail = "no diagnostic output"
+			}
+			return 0, fmt.Errorf("acquire application lock: lock creation or schedule rendezvous failed (exit %d): %s", res.ExitCode, detail)
 		}
 		// held — inspect holder + age
 		hres, err := e.T.Run(ctx, "cat "+q(e.lockPath())+" 2>/dev/null || true")
@@ -180,6 +210,22 @@ func (e *Engine) acquireLock(ctx context.Context, deployID string, force bool, l
 	return 0, fmt.Errorf("could not acquire deploy lock")
 }
 
+// atomicApplicationLockCreateCmd writes complete metadata before publishing
+// the lock path. A noclobber redirect can create an empty lock before its write
+// fails (for example on ENOSPC), which makes an infrastructure error look like
+// contention. A same-directory hard link is an atomic no-replace claim, and
+// removing the temporary name leaves the claimed inode at lockPath.
+func atomicApplicationLockCreateCmd(lockPath, value string) string {
+	tmpPattern := lockPath + ".candidate.XXXXXX"
+	return "umask 077; tmp=$(mktemp " + q(tmpPattern) + ") || exit 80; " +
+		"cleanup() { rm -f \"$tmp\" || true; }; trap cleanup 0; trap 'exit 129' 1; trap 'exit 130' 2; trap 'exit 143' 15; " +
+		"printf '%s\\n' " + q(value) + " >\"$tmp\" || exit 80; " +
+		// Unlike ln, the POSIX link utility treats its second operand as the
+		// exact new path even when that path names a directory.
+		"if link \"$tmp\" " + q(lockPath) + "; then exit 0; fi; " +
+		"{ [ -e " + q(lockPath) + " ] || [ -L " + q(lockPath) + " ]; } && exit " + strconv.Itoa(applicationLockHeldExitCode) + "; exit 80"
+}
+
 func (e *Engine) ReleaseLock(ctx context.Context) {
 	if e.lockVal == "" {
 		return