fluid-core/internal/runtime/connector.go

package runtime

import (
	"context"
	"errors"
	"fmt"
	"io"
	"net"
	"net/http"
	"strings"
	"sync"
	"time"

	"github.com/tegwick/fluid-core/internal/contract"
)

// Connector calls the adapter process serving a revision.
//
// ArchitectureBlueprint.md section 5.6 puts protocol and dependency detail
// behind this boundary. Under out-of-process attachment (ADR-0002) the adapter
// is reached over HTTP, which is what lets it be written in any language.
type Connector struct {
	client   *http.Client
	breakers sync.Map // contract.RevisionID -> *breaker
	now      func() time.Time
}

// NewConnector returns a connector with sane transport defaults.
func NewConnector() *Connector {
	return &Connector{
		client: &http.Client{
			// No client-level timeout: per-revision timeouts come from the
			// descriptor and are applied through the request context, so one
			// slow revision cannot impose its budget on another.
			Transport: &http.Transport{
				MaxIdleConns:          100,
				MaxIdleConnsPerHost:   16,
				IdleConnTimeout:       90 * time.Second,
				TLSHandshakeTimeout:   5 * time.Second,
				ExpectContinueTimeout: time.Second,
			},
		},
		now: time.Now,
	}
}

// UpstreamError describes a failed adapter call, classified for the response
// policy and for telemetry.
type UpstreamError struct {
	Kind ErrorKind
	Err  error
}

func (e *UpstreamError) Error() string { return fmt.Sprintf("%s: %v", e.Kind, e.Err) }
func (e *UpstreamError) Unwrap() error { return e.Err }

// ErrCircuitOpen is returned while a revision's breaker is open.
var ErrCircuitOpen = errors.New("circuit open")

// Call forwards a request to the adapter serving rev.
func (c *Connector) Call(ctx context.Context, rev contract.Revision, r *http.Request, body io.Reader) (*http.Response, error) {
	b := c.breakerFor(rev)
	if !b.allow(c.now()) {
		return nil, &UpstreamError{Kind: ErrorUnavailable, Err: ErrCircuitOpen}
	}

	timeout := 5 * time.Second
	if rev.Runtime.TimeoutMS != nil && *rev.Runtime.TimeoutMS > 0 {
		timeout = time.Duration(*rev.Runtime.TimeoutMS) * time.Millisecond
	}

	attempts := 1
	var backoff time.Duration
	if rev.Runtime.Retry != nil {
		if rev.Runtime.Retry.MaxAttempts != nil && *rev.Runtime.Retry.MaxAttempts > 1 {
			attempts = int(*rev.Runtime.Retry.MaxAttempts)
		}
		if rev.Runtime.Retry.BackoffMS != nil {
			backoff = time.Duration(*rev.Runtime.Retry.BackoffMS) * time.Millisecond
		}
	}

	// Only idempotent methods are retried. Replaying a POST because an adapter
	// was slow would publish the same hall-of-helix entry twice.
	if !idempotent(r.Method) {
		attempts = 1
	}

	var lastErr error
	for attempt := 0; attempt < attempts; attempt++ {
		if attempt > 0 {
			select {
			case <-ctx.Done():
				return nil, &UpstreamError{Kind: ErrorTimeout, Err: ctx.Err()}
			case <-time.After(backoff):
			}
		}

		resp, err := c.attempt(ctx, rev, r, body, timeout)
		if err == nil {
			b.record(true, c.now(), rev)
			return resp, nil
		}
		lastErr = err

		// A retry needs a rewindable body; without one, one attempt is all
		// there is.
		if body != nil {
			if seeker, ok := body.(io.Seeker); ok {
				if _, serr := seeker.Seek(0, io.SeekStart); serr != nil {
					break
				}
			} else {
				break
			}
		}
	}

	b.record(false, c.now(), rev)
	return nil, lastErr
}

func (c *Connector) attempt(ctx context.Context, rev contract.Revision, r *http.Request, body io.Reader, timeout time.Duration) (*http.Response, error) {
	callCtx, cancel := context.WithTimeout(ctx, timeout)

	target := strings.TrimSuffix(rev.Runtime.Upstream, "/") + r.URL.RequestURI()
	req, err := http.NewRequestWithContext(callCtx, r.Method, target, body)
	if err != nil {
		cancel()
		return nil, &UpstreamError{Kind: ErrorUnavailable, Err: err}
	}

	// Forward consumer headers, minus hop-by-hop ones.
	for k, vs := range r.Header {
		if hopByHop[strings.ToLower(k)] {
			continue
		}
		for _, v := range vs {
			req.Header.Add(k, v)
		}
	}
	req.Header.Set("X-FLUID-Revision", string(rev.ID))
	req.Header.Set("X-FLUID-Interface", string(rev.Interface))

	resp, err := c.client.Do(req)
	if err != nil {
		cancel()
		kind := ErrorUnavailable
		var netErr net.Error
		if errors.Is(err, context.DeadlineExceeded) || (errors.As(err, &netErr) && netErr.Timeout()) {
			kind = ErrorTimeout
		}
		return nil, &UpstreamError{Kind: kind, Err: err}
	}

	// The body is still streaming, so cancellation has to outlive this call.
	resp.Body = &cancelOnClose{ReadCloser: resp.Body, cancel: cancel}
	return resp, nil
}

func idempotent(method string) bool {
	switch method {
	case http.MethodGet, http.MethodHead, http.MethodOptions, http.MethodPut, http.MethodDelete:
		return true
	}
	return false
}

var hopByHop = map[string]bool{
	"connection": true, "keep-alive": true, "proxy-authenticate": true,
	"proxy-authorization": true, "te": true, "trailer": true,
	"transfer-encoding": true, "upgrade": true,
}

type cancelOnClose struct {
	io.ReadCloser
	cancel context.CancelFunc
	once   sync.Once
}

func (c *cancelOnClose) Close() error {
	err := c.ReadCloser.Close()
	c.once.Do(c.cancel)
	return err
}

// breaker is a per-revision circuit breaker.
//
// It is per-revision rather than per-host because two revisions frequently
// share a host: an adapter that is broken at R-3 should not take R-2 down
// with it.
type breaker struct {
	mu           sync.Mutex
	failures     int64
	threshold    int64
	openUntil    time.Time
	resetAfter   time.Duration
	probeAllowed bool
}

func (c *Connector) breakerFor(rev contract.Revision) *breaker {
	if v, ok := c.breakers.Load(rev.ID); ok {
		return v.(*breaker)
	}
	b := &breaker{threshold: 5, resetAfter: 30 * time.Second}
	if cb := rev.Runtime.CircuitBreaker; cb != nil {
		if cb.FailureThreshold != nil && *cb.FailureThreshold > 0 {
			b.threshold = *cb.FailureThreshold
		}
		if cb.ResetAfterMS != nil && *cb.ResetAfterMS > 0 {
			b.resetAfter = time.Duration(*cb.ResetAfterMS) * time.Millisecond
		}
	}
	actual, _ := c.breakers.LoadOrStore(rev.ID, b)
	return actual.(*breaker)
}

func (b *breaker) allow(now time.Time) bool {
	b.mu.Lock()
	defer b.mu.Unlock()
	if b.openUntil.IsZero() || now.After(b.openUntil) {
		if !b.openUntil.IsZero() {
			// Half-open: let exactly one request through to test recovery.
			if b.probeAllowed {
				return false
			}
			b.probeAllowed = true
		}
		return true
	}
	return false
}

func (b *breaker) record(success bool, now time.Time, rev contract.Revision) {
	b.mu.Lock()
	defer b.mu.Unlock()
	if success {
		b.failures = 0
		b.openUntil = time.Time{}
		b.probeAllowed = false
		return
	}
	b.failures++
	if b.failures >= b.threshold {
		b.openUntil = now.Add(b.resetAfter)
		b.probeAllowed = false
	}
}