1package daemon23import (4 "context"5 "errors"6 "fmt"7 "os"8 "path/filepath"9 "strings"10 "sync"11 "syscall"12 "time"1314 "github.com/FelineStateMachine/tinyrelay/internal/catalog"15)1617// ProcessLock is held for the lifetime of a serving process. Catalog-only CLI18// commands remain usable while the server owns the directory.19type ProcessLock struct{ file *os.File }2021type lifecycleState struct {22 mu sync.Mutex23 started bool24 processLock *ProcessLock25 baseURL string26 cancel context.CancelFunc27 done chan struct{}28}2930// AcquireProcessLock exclusively locks dataDir for a serving process. Lock31// acquisition fails immediately when another process holds it. The caller32// retains the lock until every tenant using the directory has closed.33func AcquireProcessLock(dataDir string) (*ProcessLock, error) {34 if dataDir == "" {35 return nil, errors.New("daemon: data directory is required")36 }37 if err := os.MkdirAll(dataDir, 0700); err != nil {38 return nil, err39 }40 file, err := os.OpenFile(filepath.Join(dataDir, ".tiny.lock"), os.O_CREATE|os.O_RDWR, 0600)41 if err != nil {42 return nil, err43 }44 if err := syscall.Flock(int(file.Fd()), syscall.LOCK_EX|syscall.LOCK_NB); err != nil {45 _ = file.Close()46 return nil, fmt.Errorf("daemon: data directory is already served or cannot be locked: %w", err)47 }48 return &ProcessLock{file: file}, nil49}5051// Close releases the process lock and closes its file handle.52func (l *ProcessLock) Close() error {53 if l == nil || l.file == nil {54 return nil55 }56 return errors.Join(syscall.Flock(int(l.file.Fd()), syscall.LOCK_UN), l.file.Close())57}5859// Start acquires the process lock, opens ready tenants and starts catalog60// reconciliation. baseURL supplies their public address. ctx bounds startup;61// App.Close ends the running lifecycle. Calls on a started App return nil.62func (a *App) Start(ctx context.Context, baseURL string) error {63 a.lifecycle.mu.Lock()64 defer a.lifecycle.mu.Unlock()65 if a.lifecycle.started {66 return nil67 }68 a.mu.Lock()69 closed := a.closed70 a.mu.Unlock()71 if closed {72 return errors.New("daemon is shutting down")73 }74 lock, err := AcquireProcessLock(a.cfg.DataDir)75 if err != nil {76 return err77 }78 baseURL = strings.TrimSuffix(baseURL, "/")79 if err := a.reconcileTenants(ctx, baseURL); err != nil {80 return errors.Join(err, lock.Close())81 }82 a.lifecycle.processLock = lock83 a.lifecycle.started = true84 a.lifecycle.baseURL = baseURL85 runCtx, cancel := context.WithCancel(context.Background())86 a.lifecycle.cancel = cancel87 done := make(chan struct{})88 a.lifecycle.done = done89 go func() {90 defer close(done)91 ticker := time.NewTicker(time.Second)92 defer ticker.Stop()93 for {94 select {95 case <-runCtx.Done():96 return97 case <-ticker.C:98 if err := a.reconcileTenants(runCtx, baseURL); err != nil && runCtx.Err() == nil {99 a.telemetry.Logger().Error("reconcile tenants", "error", err)100 }101 }102 }103 }()104 return nil105}106107func (a *App) stopLifecycle() error {108 a.lifecycle.mu.Lock()109 if !a.lifecycle.started {110 a.lifecycle.mu.Unlock()111 return nil112 }113 a.lifecycle.started = false114 done := a.lifecycle.done115 a.lifecycle.cancel()116 a.lifecycle.mu.Unlock()117 <-done118 return nil119}120121func (a *App) releaseProcessLock() error {122 a.lifecycle.mu.Lock()123 defer a.lifecycle.mu.Unlock()124 lock := a.lifecycle.processLock125 a.lifecycle.processLock = nil126 return lock.Close()127}128129func (a *App) startCreatedTenant(ctx context.Context, meta catalog.Tenant) error {130 a.lifecycle.mu.Lock()131 base, started := a.lifecycle.baseURL, a.lifecycle.started132 a.lifecycle.mu.Unlock()133 if !started || meta.Status != catalog.StatusReady {134 return nil135 }136 if meta.Name != a.cfg.DefaultTenant {137 base += "/r/" + meta.Name138 }139 _, err := a.tenant(ctx, meta, base)140 return err141}142143func (a *App) reconcileTenants(ctx context.Context, base string) error {144 metas, err := a.catalog.List(ctx)145 if err != nil {146 return err147 }148 ready := make(map[string]bool, len(metas))149 for _, meta := range metas {150 ready[meta.ID] = meta.Status == catalog.StatusReady151 }152 var stopped []*Tenant153 a.mu.Lock()154 for id, tenant := range a.tenants {155 if !ready[id] {156 delete(a.tenants, id)157 stopped = append(stopped, tenant)158 }159 }160 a.mu.Unlock()161 var errs []error162 for _, tenant := range stopped {163 errs = append(errs, tenant.Close(ctx))164 }165 for _, meta := range metas {166 if !ready[meta.ID] {167 continue168 }169 u := base170 if meta.Name != a.cfg.DefaultTenant {171 u += "/r/" + meta.Name172 }173 if _, err := a.tenant(ctx, meta, u); err != nil {174 errs = append(errs, err)175 }176 }177 return errors.Join(errs...)178}