internal/domain/decision_test.go

@@ -289,3 +289,38 @@ func TestDecisionEventValidation(t *testing.T) {
 		t.Fatalf("empty supersede payload: want ErrInvalid, got %v", err)
 	}
 }
+
+// A durable event must validate the same way forever. Comparing until_ns
+// against the current clock made every lease event fail once it expired, so
+// replaying the log after a restart refused the store's own history and the
+// coordinator could not start. Found live: the API entered a restart loop
+// logging "invalid event: until_ns required".
+func TestLeaseEventsValidateAfterTheyExpire(t *testing.T) {
+	past := float64(time.Now().Add(-24 * time.Hour).UnixNano())
+	for _, e := range []Event{
+		{ID: "a", Type: "TaskLeased", TaskID: "t", Version: 2, Surface: "system", Payload: mustPayload(map[string]any{
+			"harness_id": "h1", "until_ns": past, "expected_version": 1,
+		})},
+		{ID: "b", Type: "TaskLeaseRenewed", TaskID: "t", Version: 3, Surface: "system", Payload: mustPayload(map[string]any{
+			"harness_id": "h1", "until_ns": past, "expected_version": 2,
+		})},
+	} {
+		if err := ValidateEvent(e); err != nil {
+			t.Fatalf("%s failed validation after expiry: %v", e.Type, err)
+		}
+	}
+	// Well-formedness is still checked.
+	if err := ValidateEvent(Event{ID: "c", Type: "TaskLeaseRenewed", TaskID: "t", Version: 3, Surface: "system", Payload: mustPayload(map[string]any{
+		"harness_id": "h1", "expected_version": 2,
+	})}); err == nil {
+		t.Fatal("a renewal with no until_ns was accepted")
+	}
+}
+
+func mustPayload(v map[string]any) []byte {
+	b, err := json.Marshal(v)
+	if err != nil {
+		panic(err)
+	}
+	return b
+}

internal/domain/domain.go

@@ -299,12 +299,19 @@ func ValidatePayload(typ string, p map[string]any) error {
 		if err := requiredString("harness_id"); err != nil {
 			return err
 		}
+		// Validation of a durable event must not depend on the current clock.
+		// Comparing until_ns against time.Now() here made every lease event
+		// fail validation once it expired, so replaying the log after a
+		// restart refused the store's own history and the coordinator could
+		// not start at all. Freshness is a lease question, answered by
+		// Store.Lease and Store.ExpireLeases; well-formedness is this
+		// function's question.
 		until, untilOK := p["until_ns"].(float64)
 		if ttl, ok := p["ttl"].(float64); ok {
 			if ttl <= 0 {
 				return fmt.Errorf("%w: ttl invalid", ErrInvalid)
 			}
-		} else if !untilOK || until <= float64(time.Now().UnixNano()) {
+		} else if !untilOK || until <= 0 {
 			return fmt.Errorf("%w: ttl required", ErrInvalid)
 		}
 		if v, ok := p["expected_version"].(float64); !ok || v < 0 || v != float64(int(v)) {
@@ -314,8 +321,9 @@ func ValidatePayload(typ string, p map[string]any) error {
 		if err := requiredString("harness_id"); err != nil {
 			return err
 		}
+		// Time-independent for the same reason as TaskLeased above.
 		until, ok := p["until_ns"].(float64)
-		if !ok || until <= float64(time.Now().UnixNano()) {
+		if !ok || until <= 0 {
 			return fmt.Errorf("%w: until_ns required", ErrInvalid)
 		}
 		if v, ok := p["expected_version"].(float64); !ok || v < 0 || v != float64(int(v)) {