corago/codex-keep/run.go

package main

import (
	"context"
	"errors"
	"fmt"
	"io"
	mathrand "math/rand/v2"
	"time"
)

func jitter(d time.Duration) time.Duration {
	return time.Duration(float64(d) * (0.9 + mathrand.Float64()*0.2))
}

func run(ctx context.Context, opts options, log io.Writer) (runErr error) {
	s, err := openStore(opts.state)
	if err != nil {
		return err
	}
	defer func() { runErr = errors.Join(runErr, s.close()) }()
	failures := 0
	for ctx.Err() == nil {
		started := time.Now()
		err := session(ctx, opts, s, log)
		if ctx.Err() != nil {
			return nil
		}
		var diskErr *stateError
		if errors.As(err, &diskErr) {
			return err
		}
		if time.Since(started) >= opts.cap {
			failures = 0
		}
		failures++
		delay := min(opts.cap, jitter(backoff(opts.base, opts.cap, failures)))
		_, _ = fmt.Fprintf(log, "codex-keep: connection ended (%v); reconnecting in %s\n", err, delay.Round(time.Millisecond))
		timer := time.NewTimer(delay)
		select {
		case <-ctx.Done():
			timer.Stop()
			return nil
		case <-timer.C:
		}
	}
	return nil
}

func session(ctx context.Context, opts options, s *store, log io.Writer) error {
	c, err := connect(ctx, opts.socket)
	if err != nil {
		return err
	}
	defer c.stop()
	op := &operator{opts: opts, store: s, conn: c, log: log, attached: make(map[string]bool), jitter: jitter}
	c.onEvent = op.event
	var init initParams
	init.Client.Name, init.Client.Version = "codex-keep", "0.1.0"
	init.Capabilities.Experimental = true
	if err := c.call(ctx, "initialize", init, nil); err != nil {
		return err
	}
	if err := c.send(0, "initialized", struct{}{}); err != nil {
		return err
	}
	_, _ = fmt.Fprintln(log, "codex-keep: connected to shared daemon")
	nextPoll := time.Now()
	for {
		now := time.Now()
		if !nextPoll.After(now) {
			if err := op.reconcile(ctx); err != nil {
				return err
			}
			nextPoll = time.Now().Add(opts.poll)
		}
		wake := nextPoll
		for id := range op.attached {
			w := s.watch(id)
			if w.echo == nil && w.seed == nil && !w.Due.IsZero() && w.eligible() && w.Due.Before(wake) {
				wake = w.Due
			}
		}
		if !wake.After(time.Now()) {
			if err := op.reconcile(ctx); err != nil {
				return err
			}
			continue
		}
		timer := time.NewTimer(time.Until(wake))
		select {
		case <-ctx.Done():
			timer.Stop()
			return ctx.Err()
		case <-timer.C:
		case f := <-c.frames:
			timer.Stop()
			if f.err != nil {
				return f.err
			}
			if err := c.event(f.message); err != nil {
				return err
			}
		}
	}
}