iop/packages/go/streamgate/stream_release.go

package streamgate

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

// ErrReleaserSerializerConflict is returned when an operation conflicts
// with the current serializer state (e.g., concurrent epoch calls).
var ErrReleaserSerializerConflict = errors.New("streamgate: releaser serializer conflict")

// ErrReleaserEpochConsumed is returned when an epoch has already been fully
// confirmed.
var ErrReleaserEpochConsumed = errors.New("streamgate: releaser epoch consumed")

// ErrReleaserNoEpoch is returned when the given epoch ID does not exist.
var ErrReleaserNoEpoch = errors.New("streamgate: releaser epoch not found")

// ErrReleaserReplaceNotPermitted is returned when replace is called but the
// boundary or tail state does not permit it.
var ErrReleaserReplaceNotPermitted = errors.New("streamgate: replace not permitted")

// AttemptCanceller is the host seam for cancelling a provider attempt. The
// stream releaser calls this for idle and overflow recovery; the actual
// implementation lives in the transport adapter.
type AttemptCanceller interface {
	CancelAttempt(ctx context.Context, attemptID string) error
}

// StreamReleaser serializes EvidenceTail prepare/confirm with CommitBoundary
// release/terminal operations. It ensures zero/partial/full sink progress is
// confirmed correctly, exactly-once terminal semantics, and coordinated
// replacement.
type StreamReleaser struct {
	mu              sync.Mutex
	busy            bool
	tail            *EvidenceTail
	boundary        *CommitBoundary
	canceller       AttemptCanceller
	completedEpochs map[uint64]bool
}

// NewStreamReleaser creates a new StreamReleaser. Both tail and boundary
// must be non-nil. The canceller may be nil if FailPending is not expected
// to be called.
func NewStreamReleaser(tail *EvidenceTail, boundary *CommitBoundary, canceller AttemptCanceller) (*StreamReleaser, error) {
	if tail == nil {
		return nil, errors.New("streamgate: stream releaser tail is required")
	}
	if boundary == nil {
		return nil, errors.New("streamgate: stream releaser boundary is required")
	}
	return &StreamReleaser{
		tail:            tail,
		boundary:        boundary,
		canceller:       canceller,
		completedEpochs: make(map[uint64]bool),
	}, nil
}

func (r *StreamReleaser) isCompletedEpoch(epochID uint64) bool {
	r.mu.Lock()
	defer r.mu.Unlock()
	return r.completedEpochs != nil && r.completedEpochs[epochID]
}

func (r *StreamReleaser) markCompletedEpoch(epochID uint64) {
	r.mu.Lock()
	defer r.mu.Unlock()
	if r.completedEpochs == nil {
		r.completedEpochs = make(map[uint64]bool)
	}
	r.completedEpochs[epochID] = true
}

func (r *StreamReleaser) clearCompletedEpochs() {
	r.mu.Lock()
	defer r.mu.Unlock()
	r.completedEpochs = make(map[uint64]bool)
}

func (r *StreamReleaser) tryAcquireSerializer() error {
	r.mu.Lock()
	defer r.mu.Unlock()
	if r.busy {
		return ErrReleaserSerializerConflict
	}
	r.busy = true
	return nil
}

func (r *StreamReleaser) releaseSerializer() {
	r.mu.Lock()
	defer r.mu.Unlock()
	r.busy = false
}

func (r *StreamReleaser) releaseAndConfirm(ctx context.Context, attemptID string, epochID uint64) (ReleaseProgress, error) {
	prepared, err := r.tail.PrepareRelease(epochID)
	if err != nil {
		return ReleaseProgress{}, err
	}

	releaseEvents := prepared.ReleaseEvents()
	var (
		prog   ReleaseProgress
		relErr error
	)
	if len(releaseEvents) > 0 {
		prog, relErr = r.boundary.ReleaseSafe(ctx, attemptID, releaseEvents)
	}

	confirmErr := r.tail.ConfirmRelease(
		prepared.Token(),
		ReleaseConfirmation{ReleasedEvents: prog.ReleasedEvents()},
	)

	jointErr := errors.Join(relErr, confirmErr)
	if jointErr != nil {
		return prog, jointErr
	}

	r.markCompletedEpoch(epochID)
	return prog, nil
}

// ReleaseEpoch serializes a full release cycle: prepare from tail, release
// through boundary, and confirm sink progress back to tail.
func (r *StreamReleaser) ReleaseEpoch(ctx context.Context, attemptID string, epochID uint64) (ReleaseProgress, error) {
	if err := r.tryAcquireSerializer(); err != nil {
		return ReleaseProgress{}, err
	}
	defer r.releaseSerializer()

	return r.releaseAndConfirm(ctx, attemptID, epochID)
}

// ReleaseTerminalEpoch commits a terminal result.
func (r *StreamReleaser) ReleaseTerminalEpoch(ctx context.Context, attemptID string, epochID uint64, result TerminalResult) (ReleaseProgress, error) {
	if err := r.tryAcquireSerializer(); err != nil {
		return ReleaseProgress{}, err
	}
	defer r.releaseSerializer()

	if !result.Success() {
		return r.failPendingLocked(ctx, attemptID, result)
	}

	if r.isCompletedEpoch(epochID) {
		return ReleaseProgress{}, r.boundary.CommitTerminal(ctx, attemptID, result)
	}

	prog, err := r.releaseAndConfirm(ctx, attemptID, epochID)
	if err != nil {
		return prog, err
	}

	return prog, r.boundary.CommitTerminal(ctx, attemptID, result)
}

// ReplaceUncommittedAttempt coordinates replacing the current attempt.
func (r *StreamReleaser) ReplaceUncommittedAttempt(ctx context.Context, oldAttemptID, newAttemptID string) error {
	if err := r.tryAcquireSerializer(); err != nil {
		return err
	}
	defer r.releaseSerializer()

	if err := r.boundary.ReplaceUncommittedAttempt(oldAttemptID, newAttemptID); err != nil {
		return err
	}

	r.tail.ResetForReplace()
	r.clearCompletedEpochs()
	return nil
}

// FailPending discards all pending tail state, calls the canceller, and
// commits an error terminal through the boundary.
func (r *StreamReleaser) FailPending(ctx context.Context, attemptID string, result TerminalResult) (ReleaseProgress, error) {
	if err := r.tryAcquireSerializer(); err != nil {
		return ReleaseProgress{}, err
	}
	defer r.releaseSerializer()
	return r.failPendingLocked(ctx, attemptID, result)
}

func (r *StreamReleaser) failPendingLocked(ctx context.Context, attemptID string, result TerminalResult) (ReleaseProgress, error) {
	r.tail.DiscardPendingForTerminal()

	if r.canceller != nil {
		if err := r.canceller.CancelAttempt(ctx, attemptID); err != nil {
			if result.Error() {
				causes := result.FailureCauses()
				newCause, cerr := NewFailureCause("releaser", "attempt_cancel_failed", "", "", "")
				if cerr == nil {
					causes, _ = causes.Append(newCause)
				}
				newResult, rerr := NewErrorTerminalResult(
					result.Channel(),
					*result.ExternalDesc(),
					causes,
					result.OccurredAt(),
				)
				if rerr == nil {
					return ReleaseProgress{}, r.boundary.CommitTerminal(ctx, attemptID, newResult)
				}
			}
			return ReleaseProgress{}, r.boundary.CommitTerminal(ctx, attemptID, result)
		}
	}

	return ReleaseProgress{}, r.boundary.CommitTerminal(ctx, attemptID, result)
}

// PrepareForReplace prepares a release for the given epoch without executing.
func (r *StreamReleaser) PrepareForReplace(ctx context.Context, attemptID string, epochID uint64) (PreparedRelease, error) {
	if err := r.tryAcquireSerializer(); err != nil {
		return PreparedRelease{}, err
	}
	defer r.releaseSerializer()
	return r.tail.PrepareRelease(epochID)
}