| 1 | package control |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | |
| 7 | "reasonix/internal/event" |
| 8 | "reasonix/internal/session" |
| 9 | ) |
| 10 | |
| 11 | // admissionResult classifies what runGuarded did with a turn body. |
| 12 | type admissionResult int |
| 13 | |
| 14 | const ( |
| 15 | turnStarted admissionResult = iota |
| 16 | turnParked |
| 17 | turnDroppedRunning |
| 18 | turnDroppedRotating |
| 19 | turnDroppedClosed |
| 20 | turnDroppedDraining // generation no longer published after rebuild |
| 21 | turnDroppedWriteAuthority |
| 22 | turnDroppedAuthentication |
| 23 | ) |
| 24 | |
| 25 | // runGuarded runs body under a fresh context, guarding concurrent turns. |
| 26 | // Finishing-window arrivals park instead of dropping (see admissionResult). |
| 27 | func (c *Controller) runGuarded(body func(ctx context.Context) error) admissionResult { |
| 28 | return c.runGuardedWithAdmission(body, turnAdmission{}) |
| 29 | } |
| 30 | |
| 31 | func (c *Controller) runGuardedWithAdmission(body func(ctx context.Context) error, admission turnAdmission) admissionResult { |
| 32 | result := c.admitGuardedTurn(body, false, true, nil, nil, admission) |
| 33 | if admission.result != nil { |
| 34 | *admission.result = result |
| 35 | } |
| 36 | return result |
| 37 | } |
| 38 | |
| 39 | // runGuardedOrPark admits like runGuarded but parks the body while another |
| 40 | // turn is running instead of using the deliberately-silent running drop. |
| 41 | // Reserved for inputs that are the user's own words (the steer fallback): |
| 42 | // the FIFO drain in finishGuardedTurn delivers them the moment the current |
| 43 | // turn finishes. |
| 44 | func (c *Controller) runGuardedOrPark(body func(ctx context.Context) error) admissionResult { |
| 45 | return c.admitGuardedTurn(body, true, true, nil, nil, turnAdmission{}) |
| 46 | } |
| 47 | |
| 48 | // runGuardedInbox admits a durable item without parking it in volatile memory. |
| 49 | // onStart runs after admission is reserved and before its goroutine can finish. |
| 50 | func (c *Controller) runGuardedInbox(body func(ctx context.Context) error, onStart func()) admissionResult { |
| 51 | if !c.submissions.mu.TryLock() { |
| 52 | return turnDroppedRunning |
| 53 | } |
| 54 | defer c.submissions.mu.Unlock() |
| 55 | return c.admitGuardedTurn(body, false, false, onStart, nil, turnAdmission{}) |
| 56 | } |
| 57 | |
| 58 | func (c *Controller) runGuardedGoalRound(reservation *goalRoundReservation, body func(ctx context.Context) error) admissionResult { |
| 59 | c.submissions.mu.Lock() |
| 60 | defer c.submissions.mu.Unlock() |
| 61 | return c.admitGuardedTurn(body, false, false, nil, reservation, turnAdmission{}) |
| 62 | } |
| 63 | |
| 64 | func (c *Controller) admitGuardedTurn(body func(ctx context.Context) error, parkWhileRunning, parkWhileFinishing bool, onStart func(), goalRound *goalRoundReservation, admission turnAdmission) admissionResult { |
| 65 | if err := c.authentication.admissionError(); err != nil { |
| 66 | var authErr *AuthenticationError |
| 67 | _ = errors.As(err, &authErr) |
| 68 | code := "authentication_not_ready" |
| 69 | if authErr != nil && authErr.State.Code != "" { |
| 70 | code = authErr.State.Code |
| 71 | } |
| 72 | c.sink.Emit(event.Event{Kind: event.Notice, Level: event.LevelWarn, Code: code, Text: err.Error()}) |
| 73 | return turnDroppedAuthentication |
| 74 | } |
| 75 | // Freeze before a turn can park. Delayed execution owns this immutable |
| 76 | // admission value and never consults temporary controller state. |
| 77 | prepared := admission.images |
| 78 | admissionCtx := admission.durableCtx |
| 79 | if admissionCtx == nil { |
| 80 | admissionCtx = context.Background() |
| 81 | } |
| 82 | run := body |
| 83 | body = func(ctx context.Context) error { |
| 84 | return run(contextWithPreparedImageReferences(ctx, prepared)) |
| 85 | } |
| 86 | if err := c.ensureWriteAuthorityReady(); err != nil { |
| 87 | c.sink.Emit(event.Event{Kind: event.Notice, Level: event.LevelWarn, Text: "input was not accepted: this session is no longer writable — reopen it and try again"}) |
| 88 | return turnDroppedWriteAuthority |
| 89 | } |
| 90 | if ledger := c.turnEventLedger(); ledger != nil && ledger.CurrentStatus() == event.TurnRecoveryRequired { |
| 91 | c.sink.Emit(event.Event{Kind: event.Notice, Level: event.LevelWarn, Text: ErrRecoveryRequired.Error()}) |
| 92 | return turnDroppedWriteAuthority |
| 93 | } |
| 94 | c.mu.Lock() |
| 95 | if c.closed { |
| 96 | c.mu.Unlock() |
| 97 | return turnDroppedClosed |
| 98 | } |
| 99 | if c.rejectDrainingGenerationLocked() { |
| 100 | c.mu.Unlock() |
| 101 | c.emitDrainingNotice() |
| 102 | return turnDroppedDraining |
| 103 | } |
| 104 | if c.rotating { |
| 105 | c.mu.Unlock() |
| 106 | c.sink.Emit(event.Event{Kind: event.Notice, Level: event.LevelWarn, Text: "input was not accepted: the session is being switched — please resend"}) |
| 107 | return turnDroppedRotating |
| 108 | } |
| 109 | if c.turns.phase == session.RuntimeRecoveryRequired { |
| 110 | c.mu.Unlock() |
| 111 | c.sink.Emit(event.Event{Kind: event.Notice, Level: event.LevelWarn, Text: ErrRecoveryRequired.Error()}) |
| 112 | return turnDroppedWriteAuthority |
| 113 | } |
| 114 | kind := queuedUser |
| 115 | if goalRound != nil { |
| 116 | kind = queuedGoal |
| 117 | } |
| 118 | item := queuedTurn{kind: kind, body: body, onStart: onStart, goalRound: goalRound, admissionCtx: admissionCtx} |
| 119 | switch c.turns.phase { |
| 120 | case session.RuntimeRunning: |
| 121 | if parkWhileRunning || c.turns.cancelRequested { |
| 122 | c.queueTurnLocked(item) |
| 123 | c.mu.Unlock() |
| 124 | return turnParked |
| 125 | } |
| 126 | c.mu.Unlock() |
| 127 | return turnDroppedRunning |
| 128 | case session.RuntimeCancelling: |
| 129 | c.queueTurnLocked(item) |
| 130 | c.mu.Unlock() |
| 131 | return turnParked |
| 132 | case session.RuntimeFinalizing: |
| 133 | if !parkWhileFinishing { |
| 134 | c.mu.Unlock() |
| 135 | return turnDroppedRunning |
| 136 | } |
| 137 | c.queueTurnLocked(item) |
| 138 | c.mu.Unlock() |
| 139 | return turnParked |
| 140 | } |
| 141 | ctx, cancel, admitted := c.startTurnLocked(context.Background(), item) |
| 142 | if !admitted { |
| 143 | c.mu.Unlock() |
| 144 | c.emitDrainingNotice() |
| 145 | return turnDroppedDraining |
| 146 | } |
| 147 | c.mu.Unlock() |
| 148 | if onStart != nil { |
| 149 | onStart() |
| 150 | } |
| 151 | c.refreshRuntimeState(event.Event{}) |
| 152 | c.spawnGuardedTurn(ctx, cancel, item) |
| 153 | return turnStarted |
| 154 | } |
| 155 |