iop/packages/go/streamgate/commit_boundary.go

package streamgate

import (
	"context"
	"errors"
	"sync"
)

// ErrBoundaryUncommitted is returned when an operation requires the
// boundary to be in transport_uncommitted state but it is not.
var ErrBoundaryUncommitted = errors.New("streamgate: commit boundary not in uncommitted state")

// ErrBoundaryAlreadyTerminal is returned when a terminal has already been
// committed.
var ErrBoundaryAlreadyTerminal = errors.New("streamgate: commit boundary already terminal")

// ErrBoundaryNoStagedStart is returned when a release is attempted without
// a staged response start.
var ErrBoundaryNoStagedStart = errors.New("streamgate: no staged response start")

// ErrBoundaryDuplicateAttempt is returned when BeginAttempt is called for an
// attempt already in progress.
var ErrBoundaryDuplicateAttempt = errors.New("streamgate: duplicate attempt id")

// ErrBoundaryInvalidAttempt is returned when the given attempt ID does not
// match the current active attempt.
var ErrBoundaryInvalidAttempt = errors.New("streamgate: attempt id mismatch")

// ErrBoundaryNoCurrentAttempt is returned when no attempt has been begun.
var ErrBoundaryNoCurrentAttempt = errors.New("streamgate: no current attempt")

// ErrBoundaryReplaced is returned when an operation is attempted with an
// old attempt ID after the boundary has been replaced.
var ErrBoundaryReplaced = errors.New("streamgate: boundary replaced by newer attempt")

// ErrBoundaryRecoveryNotPermitted is returned when the current state does
// not allow the requested recovery strategy.
var ErrBoundaryRecoveryNotPermitted = errors.New("streamgate: recovery strategy not permitted for current state")

// ReleaseProgress records the number of events successfully released through
// the boundary during a ReleaseSafe call. It is immutable to callers.
type ReleaseProgress struct {
	releasedEvents int
}

// ReleasedEvents returns the count of events that the downstream sink
// accepted during this release.
func (p ReleaseProgress) ReleasedEvents() int { return p.releasedEvents }

// CommitBoundary owns the request-scope state machine for committing release
// payloads to the downstream sink. It enforces staged response-start,
// monotonic state transitions, partial progress tracking, and exactly-once
// terminal semantics.
//
// A CommitBoundary is created via NewCommitBoundary. BeginAttempt opens a new
// attempt from transport_uncommitted. StageResponseStart buffers a response
// start that is emitted at most once during the first safe release. ReleaseSafe
// commits events one at a time, tracking confirmed progress. CommitTerminal
// commits the terminal result exactly once.
//
// All public methods are safe for concurrent use. The boundary holds a mutex
// during input validation and state transitions; sink callback execution
// happens outside the lock to avoid deadlocks on State() reads.
type CommitBoundary struct {
	mu          sync.Mutex
	sink        ReleaseSink
	attemptID   string
	stagedStart *ResponseStart
	state       CommitState
	confirmed   []ReleaseEvent
}

// NewCommitBoundary creates a new CommitBoundary bound to the given sink.
// The boundary starts in transport_uncommitted state.
func NewCommitBoundary(sink ReleaseSink) (*CommitBoundary, error) {
	if sink == nil {
		return nil, errors.New("streamgate: commit boundary sink is required")
	}
	return &CommitBoundary{
		sink:  sink,
		state: CommitStateTransportUncommitted,
	}, nil
}

// BeginAttempt opens a new attempt identified by attemptID. It requires
// transport_uncommitted state and rejects if the boundary has been replaced
// (i.e., an active attempt exists that was not properly closed).
func (b *CommitBoundary) BeginAttempt(attemptID string) error {
	if attemptID == "" {
		return errors.New("streamgate: begin attempt id is required")
	}
	b.mu.Lock()
	defer b.mu.Unlock()
	if b.state != CommitStateTransportUncommitted {
		return ErrBoundaryUncommitted
	}
	if b.attemptID != "" {
		return ErrBoundaryDuplicateAttempt
	}
	b.attemptID = attemptID
	b.stagedStart = nil
	b.confirmed = nil
	return nil
}

// StageResponseStart buffers a response start for the given attempt. It
// requires transport_uncommitted state and rejects if already staged.
func (b *CommitBoundary) StageResponseStart(attemptID string, start ResponseStart) error {
	if attemptID == "" {
		return errors.New("streamgate: stage response start attempt id is required")
	}
	b.mu.Lock()
	defer b.mu.Unlock()
	if b.state != CommitStateTransportUncommitted {
		return ErrBoundaryUncommitted
	}
	if b.stagedStart != nil {
		return errors.New("streamgate: response start already staged")
	}
	b.stagedStart = &start
	return nil
}

// ReleaseSafe commits the staged response start (if any) exactly once, then
// releases each event in order. It returns the number of events successfully
// released. Events that fail at the sink are excluded from progress but do
// not regress state.
func (b *CommitBoundary) ReleaseSafe(ctx context.Context, attemptID string, events []ReleaseEvent) (ReleaseProgress, error) {
	if attemptID == "" {
		return ReleaseProgress{}, errors.New("streamgate: release safe attempt id is required")
	}
	if len(events) == 0 {
		return ReleaseProgress{}, errors.New("streamgate: release safe requires at least one event")
	}

	b.mu.Lock()

	// Validate attempt ID matches current attempt.
	if b.attemptID != "" && b.attemptID != attemptID {
		b.mu.Unlock()
		return ReleaseProgress{}, ErrBoundaryInvalidAttempt
	}

	// Snapshot staged start and current state for release outside lock.
	var rsToCommit *ResponseStart
	if b.stagedStart != nil {
		rs := *b.stagedStart
		rsToCommit = &rs
	}

	// Snapshot events to release outside lock.
	eventsCopy := make([]ReleaseEvent, len(events))
	copy(eventsCopy, events)

	// Capture current state before release.
	currentState := b.state
	b.mu.Unlock()

	// Commit staged response start outside lock if any.
	if rsToCommit != nil {
		state, err := b.sink.CommitResponseStart(ctx, *rsToCommit)
		if err != nil {
			return ReleaseProgress{}, err
		}
		currentState = state
	}

	// Release events one at a time, tracking confirmed progress.
	releaseDone := 0
	var lastErr error
	for _, ev := range eventsCopy {
		state, err := b.sink.Release(ctx, ev)
		if err != nil {
			lastErr = err
			break
		}
		releaseDone++
		currentState = state
	}

	b.mu.Lock()

	// Apply confirmed progress atomically.
	b.confirmed = append(b.confirmed, eventsCopy[:releaseDone]...)

	// If the first event was released (start or body), transition to stream_open.
	if releaseDone > 0 && b.state == CommitStateTransportUncommitted {
		b.state = CommitStateStreamOpen
	}
	// If any event transitioned to terminal, update state.
	if currentState == CommitStateTerminalCommitted {
		b.state = CommitStateTerminalCommitted
	}

	// Clear staged start since it was committed (or would have been).
	b.stagedStart = nil

	b.mu.Unlock()
	return ReleaseProgress{releasedEvents: releaseDone}, lastErr
}

// CommitTerminal commits the terminal result. It is exactly-once: only the
// first call succeeds. It accepts both success and error terminals.
//
// For uncommitted error terminals, the staged start is discarded.
// For success terminals, the start is committed first if pending, then the
// terminal is committed.
func (b *CommitBoundary) CommitTerminal(ctx context.Context, attemptID string, result TerminalResult) error {
	if attemptID == "" {
		return errors.New("streamgate: commit terminal attempt id is required")
	}

	b.mu.Lock()

	if b.state == CommitStateTerminalCommitted {
		b.mu.Unlock()
		return ErrBoundaryAlreadyTerminal
	}

	// Determine if this is an error terminal that should discard staged start.
	isError := result.Error()

	// Validate attempt ID matches.
	if b.attemptID != "" && b.attemptID != attemptID {
		b.mu.Unlock()
		return ErrBoundaryInvalidAttempt
	}

	// Snapshot data for execution outside lock.
	rsToCommit := b.stagedStart
	var rsCopy *ResponseStart
	if rsToCommit != nil && !isError {
		cp := *rsToCommit
		rsCopy = &cp
	}

	// Clear the attempt and staged start for error terminals immediately.
	if isError {
		b.stagedStart = nil
		b.attemptID = ""
	}

	b.mu.Unlock()

	// Commit staged start outside lock if not an error terminal.
	if rsCopy != nil {
		if _, err := b.sink.CommitResponseStart(ctx, *rsCopy); err != nil {
			return err
		}
	}

	// Commit terminal outside lock.
	state, err := b.sink.CommitTerminal(ctx, result)
	if err != nil {
		return err
	}

	b.mu.Lock()
	b.state = state
	if !isError {
		b.stagedStart = nil
		b.attemptID = ""
	}
	b.mu.Unlock()

	return nil
}

// ReplaceUncommittedAttempt discards the current attempt and its staged start,
// allowing a new attempt to be opened with newAttemptID via BeginAttempt.
// It requires transport_uncommitted or stream_open state with no terminal committed.
func (b *CommitBoundary) ReplaceUncommittedAttempt(oldAttemptID, newAttemptID string) error {
	if newAttemptID == "" {
		return errors.New("streamgate: replace attempt new id is required")
	}
	b.mu.Lock()
	defer b.mu.Unlock()
	if b.state == CommitStateTerminalCommitted {
		return ErrBoundaryAlreadyTerminal
	}
	if oldAttemptID != "" && b.attemptID != oldAttemptID {
		return ErrBoundaryInvalidAttempt
	}
	b.attemptID = ""
	b.stagedStart = nil
	b.confirmed = nil
	return nil
}

// CommitAllowsStrategy returns true when the current state allows the given
// recovery strategy. The matrix is:
//
//	transport_uncommitted: exact_replay, schema_repair allowed
//	stream_open:           continuation_repair allowed
//	terminal_committed:    nothing allowed
func (b *CommitBoundary) CommitAllowsStrategy(strategy RecoveryStrategy) bool {
	b.mu.Lock()
	defer b.mu.Unlock()
	switch strategy {
	case RecoveryStrategyExactReplay, RecoveryStrategySchemaRepair:
		return b.state == CommitStateTransportUncommitted
	case RecoveryStrategyContinuationRepair:
		return b.state == CommitStateStreamOpen
	default:
		return false
	}
}

// State returns the current commit state. It is safe to call concurrently
// during sink callbacks.
func (b *CommitBoundary) State() CommitState {
	b.mu.Lock()
	defer b.mu.Unlock()
	return b.state
}

// CurrentAttempt returns the current attempt ID, or "" if none.
func (b *CommitBoundary) CurrentAttempt() string {
	b.mu.Lock()
	defer b.mu.Unlock()
	return b.attemptID
}