waitinline.go

// Package waitinline provides a fair, FIFO-style locking mechanism
// where goroutines acquire access in the order they request it.
package waitinline

import (
	"context"
	"errors"
)

var (
	// ErrQueueClosed is returned when the Line has been closed.
	ErrQueueClosed = errors.New("queue closed")

	// ErrContextCancelled is returned when the provided context is cancelled before lock acquisition.
	ErrContextCancelled = errors.New("context cancelled")

	// ErrFailedUnlock is returned when unlocking fails due to improper state.
	ErrFailedUnlock = errors.New("unlock failed")
)

// LockHandler represents a ticket acquired from the line.
// It must be unlocked once done, to release the lock to the next goroutine.
type LockHandler interface {
	Unlock() error
}

type pass struct {
	signal   chan struct{}
	ctx      context.Context
	cancel   context.CancelFunc
	queuectx context.Context
}

// Line represents a fair lock where goroutines wait in order (like a queue) to acquire access.
type Line struct {
	queue  chan *pass
	ctx    context.Context
	cancel context.CancelFunc
	done   chan struct{}
}

// New returns a new Line with the specified capacity.
// Capacity determines how many tickets can wait at once.
func New(size int) *Line {
	ctx, cancel := context.WithCancel(context.Background())
	fl := &Line{
		queue:  make(chan *pass, size),
		done:   make(chan struct{}),
		ctx:    ctx,
		cancel: cancel,
	}
	go fl.serve()
	return fl
}

// serve runs the internal dispatcher which grants passes in FIFO order.
func (l *Line) serve() {
	defer close(l.done)

	for {
		select {
		case <-l.ctx.Done():
			for {
				select {
				case t := <-l.queue:
					t.cancel()
				default:
					return
				}
			}
		default:
		}

		select {
		case ticket := <-l.queue:
			select {
			case <-l.ctx.Done():
				ticket.cancel()
				continue
			case <-ticket.ctx.Done():
				continue
			default:
			}

			select {
			case ticket.signal <- struct{}{}:
				select {
				case <-l.ctx.Done():
					ticket.cancel()
					continue
				case <-ticket.ctx.Done():
					continue
				case <-ticket.signal:
					ticket.cancel()
				}
			}
		case <-l.ctx.Done():
			continue
		}
	}
}

// Close shuts down the Line and releases all waiting goroutines.
// After calling Close, no new tickets can be acquired.
func (l *Line) Close() {
	l.cancel()
	<-l.done
}

// Lock places the calling goroutine in the queue.
// It returns a LockHandler which must be unlocked after use.
// If the Line is closed or context is cancelled before acquisition, it returns an error.
func (l *Line) Lock(ctx context.Context) (LockHandler, error) {
	c, cancel := context.WithCancel(ctx)
	ticket := &pass{
		signal:   make(chan struct{}, 1),
		ctx:      c,
		cancel:   cancel,
		queuectx: l.ctx,
	}

	select {
	case <-l.ctx.Done():
		cancel()
		return nil, ErrQueueClosed
	case <-ticket.ctx.Done():
		cancel()
		return nil, ErrContextCancelled
	case l.queue <- ticket:
	}

	select {
	case <-ticket.signal:
		return ticket, nil
	case <-l.ctx.Done():
		cancel()
		return nil, ErrQueueClosed
	case <-ctx.Done():
		cancel()
		return nil, ErrContextCancelled
	}
}

// Unlock releases the ticket and allows the next goroutine in line to proceed.
// It must be called exactly once after a successful Lock acquisition.
func (p *pass) Unlock() error {
	select {
	case <-p.queuectx.Done():
		return ErrQueueClosed
	default:
	}
	select {
	case p.signal <- struct{}{}:
		return nil
	default:
		return ErrFailedUnlock
	}
}