fix(client): serialize route ledger transitions
- commit
- 15781f2dcd444fec899175c1e566aaae0d168515
- author
- brice · · 2026-08-09 15:58 UTC
- parent
- 2f86aba
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
3 files · +182 -58
M internal/usecase/client/client.go +10 -15
| @@ -528,28 +528,28 @@ func (r *Runner) Run(ctx context.Context, request AttachRequest) (retErr error) | |||
| 528 | 528 | if r.ledger == nil { | |
| 529 | 529 | return errors.New("vev: route ledger unavailable") | |
| 530 | 530 | } | |
| 531 | - | record, prior, valid, noOp := r.ledger.navigationRecords(*result.routeAction) | |
| 531 | + | selection, valid := r.ledger.navigationSelection(*result.routeAction) | |
| 532 | 532 | if !valid { | |
| 533 | 533 | return errors.New("vev: stale route navigation action") | |
| 534 | 534 | } | |
| 535 | - | if noOp { | |
| 535 | + | if selection.noOp { | |
| 536 | 536 | continue | |
| 537 | 537 | } | |
| 538 | 538 | action := *result.routeAction | |
| 539 | 539 | routeNavigationAction = &action | |
| 540 | - | routeNavigationFallback = &attachRoute{dialer: prior.dialer, request: prior.request, resumeToken: prior.resumeToken} | |
| 541 | - | dialer = record.dialer | |
| 542 | - | attemptRequest = cloneAttachRequest(record.request) | |
| 543 | - | if record.resumeToken != 0 { | |
| 540 | + | routeNavigationFallback = &attachRoute{dialer: selection.prior.dialer, request: selection.prior.request, resumeToken: selection.prior.resumeToken} | |
| 541 | + | dialer = selection.selected.dialer | |
| 542 | + | attemptRequest = cloneAttachRequest(selection.selected.request) | |
| 543 | + | if selection.selected.resumeToken != 0 { | |
| 544 | 544 | attemptRequest.Intent = ports.IntentResume | |
| 545 | 545 | } else { | |
| 546 | 546 | attemptRequest.Intent = ports.IntentAttach | |
| 547 | 547 | } | |
| 548 | 548 | attemptRequest.StartupOverlay = ports.StartupOverlayNone | |
| 549 | 549 | attemptRequest.NavigationCapabilities = 0 | |
| 550 | - | resumeToken = record.resumeToken | |
| 550 | + | resumeToken = selection.selected.resumeToken | |
| 551 | 551 | routeNavigationPending = true | |
| 552 | - | routeNavigationResumeFallback = record.resumeToken != 0 | |
| 552 | + | routeNavigationResumeFallback = selection.selected.resumeToken != 0 | |
| 553 | 553 | remote = syncReconnectRemote(reconnect, attemptRequest.Remote || r.remote) | |
| 554 | 554 | backoff = defaultReconnectBackoff.initial | |
| 555 | 555 | continue | |
| @@ -1074,8 +1074,7 @@ func (a *attachAttempt) run(ctx context.Context) attachResult { | |||
| 1074 | 1074 | if request.RemoteTarget != nil && (request.RemoteTarget.LifecycleID != committedIdentity.Target.LifecycleID || request.RemoteTarget.SessionName != committedIdentity.Target.SessionName) { | |
| 1075 | 1075 | return welcomedResult(errRouteTargetChanged) | |
| 1076 | 1076 | } | |
| 1077 | - | home := a.runner.ledger.homeRef().Key == 0 | |
| 1078 | - | if _, err := a.runner.ledger.commit(routeCandidateForAttach(request, *committedIdentity, a.dialer, resumeToken, home)); err != nil { | |
| 1077 | + | if _, err := a.runner.ledger.commitAttach(routeCandidateForAttach(request, *committedIdentity, a.dialer, resumeToken)); err != nil { | |
| 1079 | 1078 | return welcomedResult(fmt.Errorf("vev: committing route identity: %w", err)) | |
| 1080 | 1079 | } | |
| 1081 | 1080 | payload, err := ports.MarshalRecentRouteSnapshot(a.runner.ledger.snapshot()) | |
| @@ -1539,11 +1538,7 @@ func (a *attachAttempt) run(ctx context.Context) attachResult { | |||
| 1539 | 1538 | if a.runner.ledger == nil { | |
| 1540 | 1539 | return welcomedResult(errors.New("vev: route ledger unavailable")) | |
| 1541 | 1540 | } | |
| 1542 | - | active, activeOK := a.runner.ledger.lookup(a.runner.ledger.activeRef()) | |
| 1543 | - | if !activeOK { | |
| 1544 | - | return welcomedResult(errors.New("vev: committed route identity has no active route")) | |
| 1545 | - | } | |
| 1546 | - | if _, derr := a.runner.ledger.commit(routeCandidateForCommittedIdentity(active, identity)); derr != nil { | |
| 1541 | + | if _, derr := a.runner.ledger.commitCommittedIdentity(identity); derr != nil { | |
| 1547 | 1542 | return welcomedResult(fmt.Errorf("vev: committing daemon-local route identity: %w", derr)) | |
| 1548 | 1543 | } | |
| 1549 | 1544 | if derr := publishRouteSnapshot(); derr != nil { | |
| @@ -528,28 +528,28 @@ func (r *Runner) Run(ctx context.Context, request AttachRequest) (retErr error) | |||||
| 528 | if r.ledger == nil { | 528 | if r.ledger == nil { | ||
| 529 | return errors.New("vev: route ledger unavailable") | 529 | return errors.New("vev: route ledger unavailable") | ||
| 530 | } | 530 | } | ||
| 531 | - | record, prior, valid, noOp := r.ledger.navigationRecords(*result.routeAction) | 531 | + | selection, valid := r.ledger.navigationSelection(*result.routeAction) |
| 532 | if !valid { | 532 | if !valid { | ||
| 533 | return errors.New("vev: stale route navigation action") | 533 | return errors.New("vev: stale route navigation action") | ||
| 534 | } | 534 | } | ||
| 535 | - | if noOp { | 535 | + | if selection.noOp { |
| 536 | continue | 536 | continue | ||
| 537 | } | 537 | } | ||
| 538 | action := *result.routeAction | 538 | action := *result.routeAction | ||
| 539 | routeNavigationAction = &action | 539 | routeNavigationAction = &action | ||
| 540 | - | routeNavigationFallback = &attachRoute{dialer: prior.dialer, request: prior.request, resumeToken: prior.resumeToken} | 540 | + | routeNavigationFallback = &attachRoute{dialer: selection.prior.dialer, request: selection.prior.request, resumeToken: selection.prior.resumeToken} |
| 541 | - | dialer = record.dialer | 541 | + | dialer = selection.selected.dialer |
| 542 | - | attemptRequest = cloneAttachRequest(record.request) | 542 | + | attemptRequest = cloneAttachRequest(selection.selected.request) |
| 543 | - | if record.resumeToken != 0 { | 543 | + | if selection.selected.resumeToken != 0 { |
| 544 | attemptRequest.Intent = ports.IntentResume | 544 | attemptRequest.Intent = ports.IntentResume | ||
| 545 | } else { | 545 | } else { | ||
| 546 | attemptRequest.Intent = ports.IntentAttach | 546 | attemptRequest.Intent = ports.IntentAttach | ||
| 547 | } | 547 | } | ||
| 548 | attemptRequest.StartupOverlay = ports.StartupOverlayNone | 548 | attemptRequest.StartupOverlay = ports.StartupOverlayNone | ||
| 549 | attemptRequest.NavigationCapabilities = 0 | 549 | attemptRequest.NavigationCapabilities = 0 | ||
| 550 | - | resumeToken = record.resumeToken | 550 | + | resumeToken = selection.selected.resumeToken |
| 551 | routeNavigationPending = true | 551 | routeNavigationPending = true | ||
| 552 | - | routeNavigationResumeFallback = record.resumeToken != 0 | 552 | + | routeNavigationResumeFallback = selection.selected.resumeToken != 0 |
| 553 | remote = syncReconnectRemote(reconnect, attemptRequest.Remote || r.remote) | 553 | remote = syncReconnectRemote(reconnect, attemptRequest.Remote || r.remote) | ||
| 554 | backoff = defaultReconnectBackoff.initial | 554 | backoff = defaultReconnectBackoff.initial | ||
| 555 | continue | 555 | continue | ||
| @@ -1074,8 +1074,7 @@ func (a *attachAttempt) run(ctx context.Context) attachResult { | |||||
| 1074 | if request.RemoteTarget != nil && (request.RemoteTarget.LifecycleID != committedIdentity.Target.LifecycleID || request.RemoteTarget.SessionName != committedIdentity.Target.SessionName) { | 1074 | if request.RemoteTarget != nil && (request.RemoteTarget.LifecycleID != committedIdentity.Target.LifecycleID || request.RemoteTarget.SessionName != committedIdentity.Target.SessionName) { | ||
| 1075 | return welcomedResult(errRouteTargetChanged) | 1075 | return welcomedResult(errRouteTargetChanged) | ||
| 1076 | } | 1076 | } | ||
| 1077 | - | home := a.runner.ledger.homeRef().Key == 0 | 1077 | + | if _, err := a.runner.ledger.commitAttach(routeCandidateForAttach(request, *committedIdentity, a.dialer, resumeToken)); err != nil { |
| 1078 | - | if _, err := a.runner.ledger.commit(routeCandidateForAttach(request, *committedIdentity, a.dialer, resumeToken, home)); err != nil { | |||
| 1079 | return welcomedResult(fmt.Errorf("vev: committing route identity: %w", err)) | 1078 | return welcomedResult(fmt.Errorf("vev: committing route identity: %w", err)) | ||
| 1080 | } | 1079 | } | ||
| 1081 | payload, err := ports.MarshalRecentRouteSnapshot(a.runner.ledger.snapshot()) | 1080 | payload, err := ports.MarshalRecentRouteSnapshot(a.runner.ledger.snapshot()) | ||
| @@ -1539,11 +1538,7 @@ func (a *attachAttempt) run(ctx context.Context) attachResult { | |||||
| 1539 | if a.runner.ledger == nil { | 1538 | if a.runner.ledger == nil { | ||
| 1540 | return welcomedResult(errors.New("vev: route ledger unavailable")) | 1539 | return welcomedResult(errors.New("vev: route ledger unavailable")) | ||
| 1541 | } | 1540 | } | ||
| 1542 | - | active, activeOK := a.runner.ledger.lookup(a.runner.ledger.activeRef()) | 1541 | + | if _, derr := a.runner.ledger.commitCommittedIdentity(identity); derr != nil { |
| 1543 | - | if !activeOK { | |||
| 1544 | - | return welcomedResult(errors.New("vev: committed route identity has no active route")) | |||
| 1545 | - | } | |||
| 1546 | - | if _, derr := a.runner.ledger.commit(routeCandidateForCommittedIdentity(active, identity)); derr != nil { | |||
| 1547 | return welcomedResult(fmt.Errorf("vev: committing daemon-local route identity: %w", derr)) | 1542 | return welcomedResult(fmt.Errorf("vev: committing daemon-local route identity: %w", derr)) | ||
| 1548 | } | 1543 | } | ||
| 1549 | if derr := publishRouteSnapshot(); derr != nil { | 1544 | if derr := publishRouteSnapshot(); derr != nil { | ||
M internal/usecase/client/routes.go +108 -37
| @@ -105,7 +105,7 @@ func cloneCommittedIdentity(identity *ports.CommittedRouteIdentity) *ports.Commi | |||
| 105 | 105 | return © | |
| 106 | 106 | } | |
| 107 | 107 | ||
| 108 | - | func routeCandidateForAttach(request AttachRequest, identity ports.CommittedRouteIdentity, dialer ports.Dialer, resumeToken uint64, home bool) routeCandidate { | |
| 108 | + | func routeCandidateForAttach(request AttachRequest, identity ports.CommittedRouteIdentity, dialer ports.Dialer, resumeToken uint64) routeCandidate { | |
| 109 | 109 | origin := normalizeRouteOrigin(request.Origin, request.Remote) | |
| 110 | 110 | originKey := normalizeRouteOriginKey(request.OriginKey, origin) | |
| 111 | 111 | kind := ports.RouteKindLocal | |
| @@ -138,7 +138,6 @@ func routeCandidateForAttach(request AttachRequest, identity ports.CommittedRout | |||
| 138 | 138 | dialer: dialer, | |
| 139 | 139 | request: request, | |
| 140 | 140 | resumeToken: resumeToken, | |
| 141 | - | home: home, | |
| 142 | 141 | } | |
| 143 | 142 | } | |
| 144 | 143 | ||
| @@ -274,27 +273,71 @@ func (l *routeLedger) evictLocked() { | |||
| 274 | 273 | } | |
| 275 | 274 | } | |
| 276 | 275 | ||
| 277 | - | func (l *routeLedger) commit(candidate routeCandidate) (routeIdentity, error) { | |
| 276 | + | func prepareRouteCandidate(candidate routeCandidate) (routeCandidate, error) { | |
| 278 | 277 | if err := candidate.origin.Validate(); err != nil { | |
| 279 | - | return routeIdentity{}, err | |
| 278 | + | return routeCandidate{}, err | |
| 280 | 279 | } | |
| 281 | 280 | if err := candidate.target.Validate(); err != nil { | |
| 282 | - | return routeIdentity{}, err | |
| 281 | + | return routeCandidate{}, err | |
| 283 | 282 | } | |
| 284 | 283 | if err := validateRoutePresentation(candidate.presentation); err != nil { | |
| 285 | - | return routeIdentity{}, err | |
| 284 | + | return routeCandidate{}, err | |
| 286 | 285 | } | |
| 287 | 286 | candidate.originKey = normalizeRouteOriginKey(candidate.originKey, candidate.origin) | |
| 288 | 287 | candidate.request = cloneAttachRequest(candidate.request) | |
| 289 | 288 | candidate.request.Origin = candidate.origin | |
| 290 | 289 | candidate.request.OriginKey = candidate.originKey | |
| 291 | 290 | candidate.request.ExactTarget = &candidate.target | |
| 291 | + | return candidate, nil | |
| 292 | + | } | |
| 293 | + | ||
| 294 | + | func (l *routeLedger) commit(candidate routeCandidate) (routeIdentity, error) { | |
| 295 | + | candidate, err := prepareRouteCandidate(candidate) | |
| 296 | + | if err != nil { | |
| 297 | + | return routeIdentity{}, err | |
| 298 | + | } | |
| 292 | 299 | ||
| 293 | 300 | l.mu.Lock() | |
| 294 | 301 | defer l.mu.Unlock() | |
| 295 | 302 | return l.commitLocked(candidate) | |
| 296 | 303 | } | |
| 297 | 304 | ||
| 305 | + | // commitAttach assigns the process home route while holding the same lock as | |
| 306 | + | // the commit. Two concurrent initial attachments therefore cannot both claim | |
| 307 | + | // an empty home reference. | |
| 308 | + | func (l *routeLedger) commitAttach(candidate routeCandidate) (routeIdentity, error) { | |
| 309 | + | candidate, err := prepareRouteCandidate(candidate) | |
| 310 | + | if err != nil { | |
| 311 | + | return routeIdentity{}, err | |
| 312 | + | } | |
| 313 | + | ||
| 314 | + | l.mu.Lock() | |
| 315 | + | defer l.mu.Unlock() | |
| 316 | + | candidate.home = l.home.empty() | |
| 317 | + | return l.commitLocked(candidate) | |
| 318 | + | } | |
| 319 | + | ||
| 320 | + | // commitCommittedIdentity snapshots the current active record and commits the | |
| 321 | + | // daemon's local transition under one lock. The active route, origin, dialer, | |
| 322 | + | // and attach request cannot be read from different ledger generations. | |
| 323 | + | func (l *routeLedger) commitCommittedIdentity(identity ports.CommittedRouteIdentity) (routeIdentity, error) { | |
| 324 | + | if err := identity.Validate(); err != nil { | |
| 325 | + | return routeIdentity{}, err | |
| 326 | + | } | |
| 327 | + | ||
| 328 | + | l.mu.Lock() | |
| 329 | + | defer l.mu.Unlock() | |
| 330 | + | activeIndex := l.indexByIdentityLocked(l.active) | |
| 331 | + | if activeIndex < 0 { | |
| 332 | + | return routeIdentity{}, errors.New("active route unavailable") | |
| 333 | + | } | |
| 334 | + | candidate, err := prepareRouteCandidate(routeCandidateForCommittedIdentity(l.entries[activeIndex], identity)) | |
| 335 | + | if err != nil { | |
| 336 | + | return routeIdentity{}, err | |
| 337 | + | } | |
| 338 | + | return l.commitLocked(candidate) | |
| 339 | + | } | |
| 340 | + | ||
| 298 | 341 | func (l *routeLedger) commitLocked(candidate routeCandidate) (routeIdentity, error) { | |
| 299 | 342 | identity, err := l.nextIdentityLocked() | |
| 300 | 343 | if err != nil { | |
| @@ -304,6 +347,7 @@ func (l *routeLedger) commitLocked(candidate routeCandidate) (routeIdentity, err | |||
| 304 | 347 | if index >= 0 { | |
| 305 | 348 | identity.key = l.entries[index].identity.key | |
| 306 | 349 | } | |
| 350 | + | isHome := candidate.home || (l.home.key != 0 && l.home.key == identity.key) | |
| 307 | 351 | entry := routeRecord{ | |
| 308 | 352 | identity: identity, | |
| 309 | 353 | origin: candidate.origin, | |
| @@ -313,7 +357,7 @@ func (l *routeLedger) commitLocked(candidate routeCandidate) (routeIdentity, err | |||
| 313 | 357 | dialer: candidate.dialer, | |
| 314 | 358 | request: candidate.request, | |
| 315 | 359 | resumeToken: candidate.resumeToken, | |
| 316 | - | home: candidate.home, | |
| 360 | + | home: isHome, | |
| 317 | 361 | } | |
| 318 | 362 | oldActive := l.active | |
| 319 | 363 | l.moveToFrontLocked(index, entry) | |
| @@ -324,7 +368,7 @@ func (l *routeLedger) commitLocked(candidate routeCandidate) (routeIdentity, err | |||
| 324 | 368 | l.previous = routeIdentity{} | |
| 325 | 369 | } | |
| 326 | 370 | l.active = identity | |
| 327 | - | if candidate.home { | |
| 371 | + | if isHome { | |
| 328 | 372 | l.home = identity | |
| 329 | 373 | } | |
| 330 | 374 | if l.home.key == identity.key { | |
| @@ -422,11 +466,13 @@ func (l *routeLedger) homeRef() ports.RouteRef { | |||
| 422 | 466 | // preserves the original origin and exact target while leaving display labels | |
| 423 | 467 | // and transport capabilities separate. | |
| 424 | 468 | type routeTransitionPlan struct { | |
| 425 | - | identity routeIdentity | |
| 426 | - | record routeRecord | |
| 427 | - | origin ports.RouteOrigin | |
| 428 | - | originKey string | |
| 429 | - | target ports.ExactSessionTarget | |
| 469 | + | identity routeIdentity | |
| 470 | + | record routeRecord | |
| 471 | + | snapshotGeneration routeGeneration | |
| 472 | + | active routeIdentity | |
| 473 | + | origin ports.RouteOrigin | |
| 474 | + | originKey string | |
| 475 | + | target ports.ExactSessionTarget | |
| 430 | 476 | } | |
| 431 | 477 | ||
| 432 | 478 | type routeTransitionCommit struct { | |
| @@ -442,50 +488,67 @@ type routeTransitionConnector interface { | |||
| 442 | 488 | Restore(context.Context, routeRecord) error | |
| 443 | 489 | } | |
| 444 | 490 | ||
| 445 | - | // navigationRecords validates an action against the latest complete snapshot | |
| 446 | - | // and captures both the selected route and the live route that must be | |
| 447 | - | // restored if the transition fails. Capturing them under one read lock keeps a | |
| 448 | - | // concurrent committed-identity update from changing the restoration target | |
| 449 | - | // between validation and dialing. | |
| 450 | - | func (l *routeLedger) navigationRecords(action ports.RouteNavigationAction) (routeRecord, routeRecord, bool, bool) { | |
| 491 | + | type routeNavigationSelection struct { | |
| 492 | + | selected routeRecord | |
| 493 | + | prior routeRecord | |
| 494 | + | snapshotGeneration routeGeneration | |
| 495 | + | active routeIdentity | |
| 496 | + | noOp bool | |
| 497 | + | } | |
| 498 | + | ||
| 499 | + | // navigationSelection validates an action against the latest complete | |
| 500 | + | // snapshot and captures the selected route, live restoration route, snapshot | |
| 501 | + | // generation, and active identity under one read lock. The commit phase uses | |
| 502 | + | // the captured generation and active identity to reject interleavings that | |
| 503 | + | // changed the ledger while dialing. | |
| 504 | + | func (l *routeLedger) navigationSelection(action ports.RouteNavigationAction) (routeNavigationSelection, bool) { | |
| 451 | 505 | if action.SnapshotGeneration == 0 || action.Key == 0 || action.Generation == 0 { | |
| 452 | - | return routeRecord{}, routeRecord{}, false, false | |
| 506 | + | return routeNavigationSelection{}, false | |
| 453 | 507 | } | |
| 454 | 508 | identity := identityFromWire(ports.RouteRef{Key: action.Key, Generation: action.Generation}) | |
| 455 | 509 | l.mu.RLock() | |
| 456 | 510 | defer l.mu.RUnlock() | |
| 457 | 511 | if uint64(l.generation) != action.SnapshotGeneration { | |
| 458 | - | return routeRecord{}, routeRecord{}, false, false | |
| 512 | + | return routeNavigationSelection{}, false | |
| 513 | + | } | |
| 514 | + | selection := routeNavigationSelection{ | |
| 515 | + | snapshotGeneration: l.generation, | |
| 516 | + | active: l.active, | |
| 459 | 517 | } | |
| 460 | 518 | if identity == l.active { | |
| 461 | - | return routeRecord{}, routeRecord{}, true, true | |
| 519 | + | selection.noOp = true | |
| 520 | + | return selection, true | |
| 462 | 521 | } | |
| 463 | 522 | selectedIndex := l.indexByIdentityLocked(identity) | |
| 464 | 523 | activeIndex := l.indexByIdentityLocked(l.active) | |
| 465 | 524 | if selectedIndex < 0 || activeIndex < 0 { | |
| 466 | - | return routeRecord{}, routeRecord{}, false, false | |
| 525 | + | return routeNavigationSelection{}, false | |
| 467 | 526 | } | |
| 468 | - | return cloneRouteRecord(l.entries[selectedIndex]), cloneRouteRecord(l.entries[activeIndex]), true, false | |
| 527 | + | selection.selected = cloneRouteRecord(l.entries[selectedIndex]) | |
| 528 | + | selection.prior = cloneRouteRecord(l.entries[activeIndex]) | |
| 529 | + | return selection, true | |
| 469 | 530 | } | |
| 470 | 531 | ||
| 471 | 532 | // navigationRecord validates that an action belongs to the latest complete | |
| 472 | 533 | // snapshot. The active route is metadata-only and selecting it is a no-op. | |
| 473 | 534 | func (l *routeLedger) navigationRecord(action ports.RouteNavigationAction) (routeRecord, bool, bool) { | |
| 474 | - | selected, _, ok, noOp := l.navigationRecords(action) | |
| 475 | - | return selected, ok, noOp | |
| 535 | + | selection, ok := l.navigationSelection(action) | |
| 536 | + | return selection.selected, ok, selection.noOp | |
| 476 | 537 | } | |
| 477 | 538 | ||
| 478 | 539 | // navigate resolves a captured key/generation, attempts resume before exact | |
| 479 | 540 | // attach, and commits only after the connector returns a matching identity. | |
| 480 | 541 | // A failed transition restores the prior live route and leaves history alone. | |
| 481 | 542 | func (l *routeLedger) navigate(ctx context.Context, action ports.RouteNavigationAction, connector routeTransitionConnector) error { | |
| 482 | - | record, prior, ok, noOp := l.navigationRecords(action) | |
| 543 | + | selection, ok := l.navigationSelection(action) | |
| 483 | 544 | if !ok { | |
| 484 | 545 | return errRouteStaleSelection | |
| 485 | 546 | } | |
| 486 | - | if noOp { | |
| 547 | + | if selection.noOp { | |
| 487 | 548 | return nil | |
| 488 | 549 | } | |
| 550 | + | record := selection.selected | |
| 551 | + | prior := selection.prior | |
| 489 | 552 | if connector == nil { | |
| 490 | 553 | return errors.New("route transition connector is nil") | |
| 491 | 554 | } | |
| @@ -494,11 +557,13 @@ func (l *routeLedger) navigate(ctx context.Context, action ports.RouteNavigation | |||
| 494 | 557 | } | |
| 495 | 558 | selected := ports.RouteRef{Key: action.Key, Generation: action.Generation} | |
| 496 | 559 | plan := routeTransitionPlan{ | |
| 497 | - | identity: identityFromWire(selected), | |
| 498 | - | record: record, | |
| 499 | - | origin: record.origin, | |
| 500 | - | originKey: record.originKey, | |
| 501 | - | target: record.target, | |
| 560 | + | identity: identityFromWire(selected), | |
| 561 | + | record: record, | |
| 562 | + | snapshotGeneration: selection.snapshotGeneration, | |
| 563 | + | active: selection.active, | |
| 564 | + | origin: record.origin, | |
| 565 | + | originKey: record.originKey, | |
| 566 | + | target: record.target, | |
| 502 | 567 | } | |
| 503 | 568 | restoreOnFailure := func(err error) error { | |
| 504 | 569 | if restoreErr := connector.Restore(ctx, prior); restoreErr != nil { | |
| @@ -543,17 +608,23 @@ func (l *routeLedger) navigate(ctx context.Context, action ports.RouteNavigation | |||
| 543 | 608 | candidate.request.Origin = plan.origin | |
| 544 | 609 | candidate.request.OriginKey = plan.originKey | |
| 545 | 610 | candidate.request.ExactTarget = &candidate.target | |
| 546 | - | if err := l.commitTransition(plan.identity, candidate); err != nil { | |
| 611 | + | if err := l.commitTransition(plan.identity, plan.snapshotGeneration, plan.active, candidate); err != nil { | |
| 547 | 612 | return restoreOnFailure(err) | |
| 548 | 613 | } | |
| 549 | 614 | return nil | |
| 550 | 615 | } | |
| 551 | 616 | ||
| 552 | - | func (l *routeLedger) commitTransition(selected routeIdentity, candidate routeCandidate) error { | |
| 617 | + | func (l *routeLedger) commitTransition(selected routeIdentity, snapshotGeneration routeGeneration, active routeIdentity, candidate routeCandidate) error { | |
| 618 | + | candidate, err := prepareRouteCandidate(candidate) | |
| 619 | + | if err != nil { | |
| 620 | + | return err | |
| 621 | + | } | |
| 553 | 622 | l.mu.Lock() | |
| 554 | 623 | defer l.mu.Unlock() | |
| 555 | - | index := l.indexByIdentityLocked(selected) | |
| 556 | - | if index < 0 { | |
| 624 | + | if l.generation != snapshotGeneration || l.active != active { | |
| 625 | + | return errRouteStaleSelection | |
| 626 | + | } | |
| 627 | + | if l.indexByIdentityLocked(selected) < 0 { | |
| 557 | 628 | return errRouteStaleSelection | |
| 558 | 629 | } | |
| 559 | 630 | if _, err := l.commitLocked(candidate); err != nil { | |
| @@ -105,7 +105,7 @@ func cloneCommittedIdentity(identity *ports.CommittedRouteIdentity) *ports.Commi | |||||
| 105 | return © | 105 | return © | ||
| 106 | } | 106 | } | ||
| 107 | 107 | ||||
| 108 | - | func routeCandidateForAttach(request AttachRequest, identity ports.CommittedRouteIdentity, dialer ports.Dialer, resumeToken uint64, home bool) routeCandidate { | 108 | + | func routeCandidateForAttach(request AttachRequest, identity ports.CommittedRouteIdentity, dialer ports.Dialer, resumeToken uint64) routeCandidate { |
| 109 | origin := normalizeRouteOrigin(request.Origin, request.Remote) | 109 | origin := normalizeRouteOrigin(request.Origin, request.Remote) | ||
| 110 | originKey := normalizeRouteOriginKey(request.OriginKey, origin) | 110 | originKey := normalizeRouteOriginKey(request.OriginKey, origin) | ||
| 111 | kind := ports.RouteKindLocal | 111 | kind := ports.RouteKindLocal | ||
| @@ -138,7 +138,6 @@ func routeCandidateForAttach(request AttachRequest, identity ports.CommittedRout | |||||
| 138 | dialer: dialer, | 138 | dialer: dialer, | ||
| 139 | request: request, | 139 | request: request, | ||
| 140 | resumeToken: resumeToken, | 140 | resumeToken: resumeToken, | ||
| 141 | - | home: home, | |||
| 142 | } | 141 | } | ||
| 143 | } | 142 | } | ||
| 144 | 143 | ||||
| @@ -274,27 +273,71 @@ func (l *routeLedger) evictLocked() { | |||||
| 274 | } | 273 | } | ||
| 275 | } | 274 | } | ||
| 276 | 275 | ||||
| 277 | - | func (l *routeLedger) commit(candidate routeCandidate) (routeIdentity, error) { | 276 | + | func prepareRouteCandidate(candidate routeCandidate) (routeCandidate, error) { |
| 278 | if err := candidate.origin.Validate(); err != nil { | 277 | if err := candidate.origin.Validate(); err != nil { | ||
| 279 | - | return routeIdentity{}, err | 278 | + | return routeCandidate{}, err |
| 280 | } | 279 | } | ||
| 281 | if err := candidate.target.Validate(); err != nil { | 280 | if err := candidate.target.Validate(); err != nil { | ||
| 282 | - | return routeIdentity{}, err | 281 | + | return routeCandidate{}, err |
| 283 | } | 282 | } | ||
| 284 | if err := validateRoutePresentation(candidate.presentation); err != nil { | 283 | if err := validateRoutePresentation(candidate.presentation); err != nil { | ||
| 285 | - | return routeIdentity{}, err | 284 | + | return routeCandidate{}, err |
| 286 | } | 285 | } | ||
| 287 | candidate.originKey = normalizeRouteOriginKey(candidate.originKey, candidate.origin) | 286 | candidate.originKey = normalizeRouteOriginKey(candidate.originKey, candidate.origin) | ||
| 288 | candidate.request = cloneAttachRequest(candidate.request) | 287 | candidate.request = cloneAttachRequest(candidate.request) | ||
| 289 | candidate.request.Origin = candidate.origin | 288 | candidate.request.Origin = candidate.origin | ||
| 290 | candidate.request.OriginKey = candidate.originKey | 289 | candidate.request.OriginKey = candidate.originKey | ||
| 291 | candidate.request.ExactTarget = &candidate.target | 290 | candidate.request.ExactTarget = &candidate.target | ||
| 291 | + | return candidate, nil | |||
| 292 | + | } | |||
| 293 | + | ||||
| 294 | + | func (l *routeLedger) commit(candidate routeCandidate) (routeIdentity, error) { | |||
| 295 | + | candidate, err := prepareRouteCandidate(candidate) | |||
| 296 | + | if err != nil { | |||
| 297 | + | return routeIdentity{}, err | |||
| 298 | + | } | |||
| 292 | 299 | ||||
| 293 | l.mu.Lock() | 300 | l.mu.Lock() | ||
| 294 | defer l.mu.Unlock() | 301 | defer l.mu.Unlock() | ||
| 295 | return l.commitLocked(candidate) | 302 | return l.commitLocked(candidate) | ||
| 296 | } | 303 | } | ||
| 297 | 304 | ||||
| 305 | + | // commitAttach assigns the process home route while holding the same lock as | |||
| 306 | + | // the commit. Two concurrent initial attachments therefore cannot both claim | |||
| 307 | + | // an empty home reference. | |||
| 308 | + | func (l *routeLedger) commitAttach(candidate routeCandidate) (routeIdentity, error) { | |||
| 309 | + | candidate, err := prepareRouteCandidate(candidate) | |||
| 310 | + | if err != nil { | |||
| 311 | + | return routeIdentity{}, err | |||
| 312 | + | } | |||
| 313 | + | ||||
| 314 | + | l.mu.Lock() | |||
| 315 | + | defer l.mu.Unlock() | |||
| 316 | + | candidate.home = l.home.empty() | |||
| 317 | + | return l.commitLocked(candidate) | |||
| 318 | + | } | |||
| 319 | + | ||||
| 320 | + | // commitCommittedIdentity snapshots the current active record and commits the | |||
| 321 | + | // daemon's local transition under one lock. The active route, origin, dialer, | |||
| 322 | + | // and attach request cannot be read from different ledger generations. | |||
| 323 | + | func (l *routeLedger) commitCommittedIdentity(identity ports.CommittedRouteIdentity) (routeIdentity, error) { | |||
| 324 | + | if err := identity.Validate(); err != nil { | |||
| 325 | + | return routeIdentity{}, err | |||
| 326 | + | } | |||
| 327 | + | ||||
| 328 | + | l.mu.Lock() | |||
| 329 | + | defer l.mu.Unlock() | |||
| 330 | + | activeIndex := l.indexByIdentityLocked(l.active) | |||
| 331 | + | if activeIndex < 0 { | |||
| 332 | + | return routeIdentity{}, errors.New("active route unavailable") | |||
| 333 | + | } | |||
| 334 | + | candidate, err := prepareRouteCandidate(routeCandidateForCommittedIdentity(l.entries[activeIndex], identity)) | |||
| 335 | + | if err != nil { | |||
| 336 | + | return routeIdentity{}, err | |||
| 337 | + | } | |||
| 338 | + | return l.commitLocked(candidate) | |||
| 339 | + | } | |||
| 340 | + | ||||
| 298 | func (l *routeLedger) commitLocked(candidate routeCandidate) (routeIdentity, error) { | 341 | func (l *routeLedger) commitLocked(candidate routeCandidate) (routeIdentity, error) { | ||
| 299 | identity, err := l.nextIdentityLocked() | 342 | identity, err := l.nextIdentityLocked() | ||
| 300 | if err != nil { | 343 | if err != nil { | ||
| @@ -304,6 +347,7 @@ func (l *routeLedger) commitLocked(candidate routeCandidate) (routeIdentity, err | |||||
| 304 | if index >= 0 { | 347 | if index >= 0 { | ||
| 305 | identity.key = l.entries[index].identity.key | 348 | identity.key = l.entries[index].identity.key | ||
| 306 | } | 349 | } | ||
| 350 | + | isHome := candidate.home || (l.home.key != 0 && l.home.key == identity.key) | |||
| 307 | entry := routeRecord{ | 351 | entry := routeRecord{ | ||
| 308 | identity: identity, | 352 | identity: identity, | ||
| 309 | origin: candidate.origin, | 353 | origin: candidate.origin, | ||
| @@ -313,7 +357,7 @@ func (l *routeLedger) commitLocked(candidate routeCandidate) (routeIdentity, err | |||||
| 313 | dialer: candidate.dialer, | 357 | dialer: candidate.dialer, | ||
| 314 | request: candidate.request, | 358 | request: candidate.request, | ||
| 315 | resumeToken: candidate.resumeToken, | 359 | resumeToken: candidate.resumeToken, | ||
| 316 | - | home: candidate.home, | 360 | + | home: isHome, |
| 317 | } | 361 | } | ||
| 318 | oldActive := l.active | 362 | oldActive := l.active | ||
| 319 | l.moveToFrontLocked(index, entry) | 363 | l.moveToFrontLocked(index, entry) | ||
| @@ -324,7 +368,7 @@ func (l *routeLedger) commitLocked(candidate routeCandidate) (routeIdentity, err | |||||
| 324 | l.previous = routeIdentity{} | 368 | l.previous = routeIdentity{} | ||
| 325 | } | 369 | } | ||
| 326 | l.active = identity | 370 | l.active = identity | ||
| 327 | - | if candidate.home { | 371 | + | if isHome { |
| 328 | l.home = identity | 372 | l.home = identity | ||
| 329 | } | 373 | } | ||
| 330 | if l.home.key == identity.key { | 374 | if l.home.key == identity.key { | ||
| @@ -422,11 +466,13 @@ func (l *routeLedger) homeRef() ports.RouteRef { | |||||
| 422 | // preserves the original origin and exact target while leaving display labels | 466 | // preserves the original origin and exact target while leaving display labels | ||
| 423 | // and transport capabilities separate. | 467 | // and transport capabilities separate. | ||
| 424 | type routeTransitionPlan struct { | 468 | type routeTransitionPlan struct { | ||
| 425 | - | identity routeIdentity | 469 | + | identity routeIdentity |
| 426 | - | record routeRecord | 470 | + | record routeRecord |
| 427 | - | origin ports.RouteOrigin | 471 | + | snapshotGeneration routeGeneration |
| 428 | - | originKey string | 472 | + | active routeIdentity |
| 429 | - | target ports.ExactSessionTarget | 473 | + | origin ports.RouteOrigin |
| 474 | + | originKey string | |||
| 475 | + | target ports.ExactSessionTarget | |||
| 430 | } | 476 | } | ||
| 431 | 477 | ||||
| 432 | type routeTransitionCommit struct { | 478 | type routeTransitionCommit struct { | ||
| @@ -442,50 +488,67 @@ type routeTransitionConnector interface { | |||||
| 442 | Restore(context.Context, routeRecord) error | 488 | Restore(context.Context, routeRecord) error | ||
| 443 | } | 489 | } | ||
| 444 | 490 | ||||
| 445 | - | // navigationRecords validates an action against the latest complete snapshot | 491 | + | type routeNavigationSelection struct { |
| 446 | - | // and captures both the selected route and the live route that must be | 492 | + | selected routeRecord |
| 447 | - | // restored if the transition fails. Capturing them under one read lock keeps a | 493 | + | prior routeRecord |
| 448 | - | // concurrent committed-identity update from changing the restoration target | 494 | + | snapshotGeneration routeGeneration |
| 449 | - | // between validation and dialing. | 495 | + | active routeIdentity |
| 450 | - | func (l *routeLedger) navigationRecords(action ports.RouteNavigationAction) (routeRecord, routeRecord, bool, bool) { | 496 | + | noOp bool |
| 497 | + | } | |||
| 498 | + | ||||
| 499 | + | // navigationSelection validates an action against the latest complete | |||
| 500 | + | // snapshot and captures the selected route, live restoration route, snapshot | |||
| 501 | + | // generation, and active identity under one read lock. The commit phase uses | |||
| 502 | + | // the captured generation and active identity to reject interleavings that | |||
| 503 | + | // changed the ledger while dialing. | |||
| 504 | + | func (l *routeLedger) navigationSelection(action ports.RouteNavigationAction) (routeNavigationSelection, bool) { | |||
| 451 | if action.SnapshotGeneration == 0 || action.Key == 0 || action.Generation == 0 { | 505 | if action.SnapshotGeneration == 0 || action.Key == 0 || action.Generation == 0 { | ||
| 452 | - | return routeRecord{}, routeRecord{}, false, false | 506 | + | return routeNavigationSelection{}, false |
| 453 | } | 507 | } | ||
| 454 | identity := identityFromWire(ports.RouteRef{Key: action.Key, Generation: action.Generation}) | 508 | identity := identityFromWire(ports.RouteRef{Key: action.Key, Generation: action.Generation}) | ||
| 455 | l.mu.RLock() | 509 | l.mu.RLock() | ||
| 456 | defer l.mu.RUnlock() | 510 | defer l.mu.RUnlock() | ||
| 457 | if uint64(l.generation) != action.SnapshotGeneration { | 511 | if uint64(l.generation) != action.SnapshotGeneration { | ||
| 458 | - | return routeRecord{}, routeRecord{}, false, false | 512 | + | return routeNavigationSelection{}, false |
| 513 | + | } | |||
| 514 | + | selection := routeNavigationSelection{ | |||
| 515 | + | snapshotGeneration: l.generation, | |||
| 516 | + | active: l.active, | |||
| 459 | } | 517 | } | ||
| 460 | if identity == l.active { | 518 | if identity == l.active { | ||
| 461 | - | return routeRecord{}, routeRecord{}, true, true | 519 | + | selection.noOp = true |
| 520 | + | return selection, true | |||
| 462 | } | 521 | } | ||
| 463 | selectedIndex := l.indexByIdentityLocked(identity) | 522 | selectedIndex := l.indexByIdentityLocked(identity) | ||
| 464 | activeIndex := l.indexByIdentityLocked(l.active) | 523 | activeIndex := l.indexByIdentityLocked(l.active) | ||
| 465 | if selectedIndex < 0 || activeIndex < 0 { | 524 | if selectedIndex < 0 || activeIndex < 0 { | ||
| 466 | - | return routeRecord{}, routeRecord{}, false, false | 525 | + | return routeNavigationSelection{}, false |
| 467 | } | 526 | } | ||
| 468 | - | return cloneRouteRecord(l.entries[selectedIndex]), cloneRouteRecord(l.entries[activeIndex]), true, false | 527 | + | selection.selected = cloneRouteRecord(l.entries[selectedIndex]) |
| 528 | + | selection.prior = cloneRouteRecord(l.entries[activeIndex]) | |||
| 529 | + | return selection, true | |||
| 469 | } | 530 | } | ||
| 470 | 531 | ||||
| 471 | // navigationRecord validates that an action belongs to the latest complete | 532 | // navigationRecord validates that an action belongs to the latest complete | ||
| 472 | // snapshot. The active route is metadata-only and selecting it is a no-op. | 533 | // snapshot. The active route is metadata-only and selecting it is a no-op. | ||
| 473 | func (l *routeLedger) navigationRecord(action ports.RouteNavigationAction) (routeRecord, bool, bool) { | 534 | func (l *routeLedger) navigationRecord(action ports.RouteNavigationAction) (routeRecord, bool, bool) { | ||
| 474 | - | selected, _, ok, noOp := l.navigationRecords(action) | 535 | + | selection, ok := l.navigationSelection(action) |
| 475 | - | return selected, ok, noOp | 536 | + | return selection.selected, ok, selection.noOp |
| 476 | } | 537 | } | ||
| 477 | 538 | ||||
| 478 | // navigate resolves a captured key/generation, attempts resume before exact | 539 | // navigate resolves a captured key/generation, attempts resume before exact | ||
| 479 | // attach, and commits only after the connector returns a matching identity. | 540 | // attach, and commits only after the connector returns a matching identity. | ||
| 480 | // A failed transition restores the prior live route and leaves history alone. | 541 | // A failed transition restores the prior live route and leaves history alone. | ||
| 481 | func (l *routeLedger) navigate(ctx context.Context, action ports.RouteNavigationAction, connector routeTransitionConnector) error { | 542 | func (l *routeLedger) navigate(ctx context.Context, action ports.RouteNavigationAction, connector routeTransitionConnector) error { | ||
| 482 | - | record, prior, ok, noOp := l.navigationRecords(action) | 543 | + | selection, ok := l.navigationSelection(action) |
| 483 | if !ok { | 544 | if !ok { | ||
| 484 | return errRouteStaleSelection | 545 | return errRouteStaleSelection | ||
| 485 | } | 546 | } | ||
| 486 | - | if noOp { | 547 | + | if selection.noOp { |
| 487 | return nil | 548 | return nil | ||
| 488 | } | 549 | } | ||
| 550 | + | record := selection.selected | |||
| 551 | + | prior := selection.prior | |||
| 489 | if connector == nil { | 552 | if connector == nil { | ||
| 490 | return errors.New("route transition connector is nil") | 553 | return errors.New("route transition connector is nil") | ||
| 491 | } | 554 | } | ||
| @@ -494,11 +557,13 @@ func (l *routeLedger) navigate(ctx context.Context, action ports.RouteNavigation | |||||
| 494 | } | 557 | } | ||
| 495 | selected := ports.RouteRef{Key: action.Key, Generation: action.Generation} | 558 | selected := ports.RouteRef{Key: action.Key, Generation: action.Generation} | ||
| 496 | plan := routeTransitionPlan{ | 559 | plan := routeTransitionPlan{ | ||
| 497 | - | identity: identityFromWire(selected), | 560 | + | identity: identityFromWire(selected), |
| 498 | - | record: record, | 561 | + | record: record, |
| 499 | - | origin: record.origin, | 562 | + | snapshotGeneration: selection.snapshotGeneration, |
| 500 | - | originKey: record.originKey, | 563 | + | active: selection.active, |
| 501 | - | target: record.target, | 564 | + | origin: record.origin, |
| 565 | + | originKey: record.originKey, | |||
| 566 | + | target: record.target, | |||
| 502 | } | 567 | } | ||
| 503 | restoreOnFailure := func(err error) error { | 568 | restoreOnFailure := func(err error) error { | ||
| 504 | if restoreErr := connector.Restore(ctx, prior); restoreErr != nil { | 569 | if restoreErr := connector.Restore(ctx, prior); restoreErr != nil { | ||
| @@ -543,17 +608,23 @@ func (l *routeLedger) navigate(ctx context.Context, action ports.RouteNavigation | |||||
| 543 | candidate.request.Origin = plan.origin | 608 | candidate.request.Origin = plan.origin | ||
| 544 | candidate.request.OriginKey = plan.originKey | 609 | candidate.request.OriginKey = plan.originKey | ||
| 545 | candidate.request.ExactTarget = &candidate.target | 610 | candidate.request.ExactTarget = &candidate.target | ||
| 546 | - | if err := l.commitTransition(plan.identity, candidate); err != nil { | 611 | + | if err := l.commitTransition(plan.identity, plan.snapshotGeneration, plan.active, candidate); err != nil { |
| 547 | return restoreOnFailure(err) | 612 | return restoreOnFailure(err) | ||
| 548 | } | 613 | } | ||
| 549 | return nil | 614 | return nil | ||
| 550 | } | 615 | } | ||
| 551 | 616 | ||||
| 552 | - | func (l *routeLedger) commitTransition(selected routeIdentity, candidate routeCandidate) error { | 617 | + | func (l *routeLedger) commitTransition(selected routeIdentity, snapshotGeneration routeGeneration, active routeIdentity, candidate routeCandidate) error { |
| 618 | + | candidate, err := prepareRouteCandidate(candidate) | |||
| 619 | + | if err != nil { | |||
| 620 | + | return err | |||
| 621 | + | } | |||
| 553 | l.mu.Lock() | 622 | l.mu.Lock() | ||
| 554 | defer l.mu.Unlock() | 623 | defer l.mu.Unlock() | ||
| 555 | - | index := l.indexByIdentityLocked(selected) | 624 | + | if l.generation != snapshotGeneration || l.active != active { |
| 556 | - | if index < 0 { | 625 | + | return errRouteStaleSelection |
| 626 | + | } | |||
| 627 | + | if l.indexByIdentityLocked(selected) < 0 { | |||
| 557 | return errRouteStaleSelection | 628 | return errRouteStaleSelection | ||
| 558 | } | 629 | } | ||
| 559 | if _, err := l.commitLocked(candidate); err != nil { | 630 | if _, err := l.commitLocked(candidate); err != nil { | ||
M internal/usecase/client/routes_test.go +64 -6
| @@ -75,18 +75,76 @@ func TestCommittedIdentityDoesNotReassignHomeRoute(t *testing.T) { | |||
| 75 | 75 | home.home = true | |
| 76 | 76 | homeIdentity, err := ledger.commit(home) | |
| 77 | 77 | require.NoError(t, err) | |
| 78 | - | active, ok := ledger.lookup(ledger.activeRef()) | |
| 79 | - | require.True(t, ok) | |
| 80 | - | ||
| 81 | - | candidate := routeCandidateForCommittedIdentity(active, ports.CommittedRouteIdentity{ | |
| 78 | + | committed, err := ledger.commitCommittedIdentity(ports.CommittedRouteIdentity{ | |
| 82 | 79 | Target: routeTestTarget(2), | |
| 83 | 80 | Ephemeral: true, | |
| 84 | 81 | }) | |
| 85 | - | committed, err := ledger.commit(candidate) | |
| 86 | 82 | require.NoError(t, err) | |
| 87 | 83 | require.Equal(t, uint64(homeIdentity.key), uint64(ledger.homeRef().Key)) | |
| 88 | 84 | require.NotEqual(t, ledger.homeRef(), committed.wire()) | |
| 89 | - | require.False(t, candidate.home) | |
| 85 | + | active, ok := ledger.lookup(ledger.activeRef()) | |
| 86 | + | require.True(t, ok) | |
| 87 | + | require.Equal(t, committed, active.identity) | |
| 88 | + | ||
| 89 | + | _, err = ledger.commit(routeTestCandidate(1, ports.RouteOriginLocal)) | |
| 90 | + | require.NoError(t, err) | |
| 91 | + | require.True(t, mustRouteRecord(t, ledger, ledger.activeRef()).home, "upserting the home route must retain its marker") | |
| 92 | + | } | |
| 93 | + | ||
| 94 | + | func TestRouteLedgerConcurrentInitialAttachmentsKeepOneHome(t *testing.T) { | |
| 95 | + | ledger := newRouteLedger() | |
| 96 | + | const attachments = 16 | |
| 97 | + | errs := make(chan error, attachments) | |
| 98 | + | var wg sync.WaitGroup | |
| 99 | + | for i := byte(0); i < attachments; i++ { | |
| 100 | + | i := i | |
| 101 | + | wg.Add(1) | |
| 102 | + | go func() { | |
| 103 | + | defer wg.Done() | |
| 104 | + | _, err := ledger.commitAttach(routeTestCandidate(i, ports.RouteOriginLocal)) | |
| 105 | + | errs <- err | |
| 106 | + | }() | |
| 107 | + | } | |
| 108 | + | wg.Wait() | |
| 109 | + | close(errs) | |
| 110 | + | for err := range errs { | |
| 111 | + | require.NoError(t, err) | |
| 112 | + | } | |
| 113 | + | ||
| 114 | + | ledger.mu.RLock() | |
| 115 | + | defer ledger.mu.RUnlock() | |
| 116 | + | homeCount := 0 | |
| 117 | + | for _, entry := range ledger.entries { | |
| 118 | + | if entry.home { | |
| 119 | + | homeCount++ | |
| 120 | + | require.Equal(t, entry.identity, ledger.home) | |
| 121 | + | } | |
| 122 | + | } | |
| 123 | + | require.Equal(t, 1, homeCount) | |
| 124 | + | } | |
| 125 | + | ||
| 126 | + | func TestRouteLedgerTransitionRejectsConcurrentCommit(t *testing.T) { | |
| 127 | + | ledger := newRouteLedger() | |
| 128 | + | selected, err := ledger.commit(routeTestCandidate(1, ports.RouteOriginLocal)) | |
| 129 | + | require.NoError(t, err) | |
| 130 | + | _, err = ledger.commit(routeTestCandidate(2, ports.RouteOriginLocal)) | |
| 131 | + | require.NoError(t, err) | |
| 132 | + | selection, ok := ledger.navigationSelection(selected.wireNavigationAction(routeGeneration(ledger.snapshot().Generation))) | |
| 133 | + | require.True(t, ok) | |
| 134 | + | ||
| 135 | + | _, err = ledger.commit(routeTestCandidate(3, ports.RouteOriginLocal)) | |
| 136 | + | require.NoError(t, err) | |
| 137 | + | candidate := routeTestCandidate(4, ports.RouteOriginLocal) | |
| 138 | + | err = ledger.commitTransition(selected, selection.snapshotGeneration, selection.active, candidate) | |
| 139 | + | require.ErrorIs(t, err, errRouteStaleSelection) | |
| 140 | + | require.Equal(t, routeTestTarget(3), mustRouteRecord(t, ledger, ledger.activeRef()).target) | |
| 141 | + | } | |
| 142 | + | ||
| 143 | + | func mustRouteRecord(t *testing.T, ledger *routeLedger, ref ports.RouteRef) routeRecord { | |
| 144 | + | t.Helper() | |
| 145 | + | record, ok := ledger.lookup(ref) | |
| 146 | + | require.True(t, ok) | |
| 147 | + | return record | |
| 90 | 148 | } | |
| 91 | 149 | ||
| 92 | 150 | func TestRouteLedgerUpsertChangesGenerationWithoutChangingOpaqueKey(t *testing.T) { | |
| @@ -75,18 +75,76 @@ func TestCommittedIdentityDoesNotReassignHomeRoute(t *testing.T) { | |||||
| 75 | home.home = true | 75 | home.home = true | ||
| 76 | homeIdentity, err := ledger.commit(home) | 76 | homeIdentity, err := ledger.commit(home) | ||
| 77 | require.NoError(t, err) | 77 | require.NoError(t, err) | ||
| 78 | - | active, ok := ledger.lookup(ledger.activeRef()) | 78 | + | committed, err := ledger.commitCommittedIdentity(ports.CommittedRouteIdentity{ |
| 79 | - | require.True(t, ok) | |||
| 80 | - | ||||
| 81 | - | candidate := routeCandidateForCommittedIdentity(active, ports.CommittedRouteIdentity{ | |||
| 82 | Target: routeTestTarget(2), | 79 | Target: routeTestTarget(2), | ||
| 83 | Ephemeral: true, | 80 | Ephemeral: true, | ||
| 84 | }) | 81 | }) | ||
| 85 | - | committed, err := ledger.commit(candidate) | |||
| 86 | require.NoError(t, err) | 82 | require.NoError(t, err) | ||
| 87 | require.Equal(t, uint64(homeIdentity.key), uint64(ledger.homeRef().Key)) | 83 | require.Equal(t, uint64(homeIdentity.key), uint64(ledger.homeRef().Key)) | ||
| 88 | require.NotEqual(t, ledger.homeRef(), committed.wire()) | 84 | require.NotEqual(t, ledger.homeRef(), committed.wire()) | ||
| 89 | - | require.False(t, candidate.home) | 85 | + | active, ok := ledger.lookup(ledger.activeRef()) |
| 86 | + | require.True(t, ok) | |||
| 87 | + | require.Equal(t, committed, active.identity) | |||
| 88 | + | ||||
| 89 | + | _, err = ledger.commit(routeTestCandidate(1, ports.RouteOriginLocal)) | |||
| 90 | + | require.NoError(t, err) | |||
| 91 | + | require.True(t, mustRouteRecord(t, ledger, ledger.activeRef()).home, "upserting the home route must retain its marker") | |||
| 92 | + | } | |||
| 93 | + | ||||
| 94 | + | func TestRouteLedgerConcurrentInitialAttachmentsKeepOneHome(t *testing.T) { | |||
| 95 | + | ledger := newRouteLedger() | |||
| 96 | + | const attachments = 16 | |||
| 97 | + | errs := make(chan error, attachments) | |||
| 98 | + | var wg sync.WaitGroup | |||
| 99 | + | for i := byte(0); i < attachments; i++ { | |||
| 100 | + | i := i | |||
| 101 | + | wg.Add(1) | |||
| 102 | + | go func() { | |||
| 103 | + | defer wg.Done() | |||
| 104 | + | _, err := ledger.commitAttach(routeTestCandidate(i, ports.RouteOriginLocal)) | |||
| 105 | + | errs <- err | |||
| 106 | + | }() | |||
| 107 | + | } | |||
| 108 | + | wg.Wait() | |||
| 109 | + | close(errs) | |||
| 110 | + | for err := range errs { | |||
| 111 | + | require.NoError(t, err) | |||
| 112 | + | } | |||
| 113 | + | ||||
| 114 | + | ledger.mu.RLock() | |||
| 115 | + | defer ledger.mu.RUnlock() | |||
| 116 | + | homeCount := 0 | |||
| 117 | + | for _, entry := range ledger.entries { | |||
| 118 | + | if entry.home { | |||
| 119 | + | homeCount++ | |||
| 120 | + | require.Equal(t, entry.identity, ledger.home) | |||
| 121 | + | } | |||
| 122 | + | } | |||
| 123 | + | require.Equal(t, 1, homeCount) | |||
| 124 | + | } | |||
| 125 | + | ||||
| 126 | + | func TestRouteLedgerTransitionRejectsConcurrentCommit(t *testing.T) { | |||
| 127 | + | ledger := newRouteLedger() | |||
| 128 | + | selected, err := ledger.commit(routeTestCandidate(1, ports.RouteOriginLocal)) | |||
| 129 | + | require.NoError(t, err) | |||
| 130 | + | _, err = ledger.commit(routeTestCandidate(2, ports.RouteOriginLocal)) | |||
| 131 | + | require.NoError(t, err) | |||
| 132 | + | selection, ok := ledger.navigationSelection(selected.wireNavigationAction(routeGeneration(ledger.snapshot().Generation))) | |||
| 133 | + | require.True(t, ok) | |||
| 134 | + | ||||
| 135 | + | _, err = ledger.commit(routeTestCandidate(3, ports.RouteOriginLocal)) | |||
| 136 | + | require.NoError(t, err) | |||
| 137 | + | candidate := routeTestCandidate(4, ports.RouteOriginLocal) | |||
| 138 | + | err = ledger.commitTransition(selected, selection.snapshotGeneration, selection.active, candidate) | |||
| 139 | + | require.ErrorIs(t, err, errRouteStaleSelection) | |||
| 140 | + | require.Equal(t, routeTestTarget(3), mustRouteRecord(t, ledger, ledger.activeRef()).target) | |||
| 141 | + | } | |||
| 142 | + | ||||
| 143 | + | func mustRouteRecord(t *testing.T, ledger *routeLedger, ref ports.RouteRef) routeRecord { | |||
| 144 | + | t.Helper() | |||
| 145 | + | record, ok := ledger.lookup(ref) | |||
| 146 | + | require.True(t, ok) | |||
| 147 | + | return record | |||
| 90 | } | 148 | } | ||
| 91 | 149 | ||||
| 92 | func TestRouteLedgerUpsertChangesGenerationWithoutChangingOpaqueKey(t *testing.T) { | 150 | func TestRouteLedgerUpsertChangesGenerationWithoutChangingOpaqueKey(t *testing.T) { | ||