| 1 | package daemon |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "fmt" |
| 7 | "os" |
| 8 | "path/filepath" |
| 9 | "strings" |
| 10 | "sync" |
| 11 | "syscall" |
| 12 | "time" |
| 13 | |
| 14 | "github.com/FelineStateMachine/tinyrelay/internal/catalog" |
| 15 | ) |
| 16 | |
| 17 | // ProcessLock is held for the lifetime of a serving process. Catalog-only CLI |
| 18 | // commands remain usable while the server owns the directory. |
| 19 | type ProcessLock struct{ file *os.File } |
| 20 | |
| 21 | type lifecycleState struct { |
| 22 | mu sync.Mutex |
| 23 | started bool |
| 24 | processLock *ProcessLock |
| 25 | baseURL string |
| 26 | cancel context.CancelFunc |
| 27 | done chan struct{} |
| 28 | } |
| 29 | |
| 30 | // AcquireProcessLock exclusively locks dataDir for a serving process. Lock |
| 31 | // acquisition fails immediately when another process holds it. The caller |
| 32 | // retains the lock until every tenant using the directory has closed. |
| 33 | func 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, err |
| 39 | } |
| 40 | file, err := os.OpenFile(filepath.Join(dataDir, ".tiny.lock" ), os.O_CREATE|os.O_RDWR, 0600) |
| 41 | if err != nil { |
| 42 | return nil, err |
| 43 | } |
| 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}, nil |
| 49 | } |
| 50 | |
| 51 | // Close releases the process lock and closes its file handle. |
| 52 | func (l *ProcessLock) Close() error { |
| 53 | if l == nil || l.file == nil { |
| 54 | return nil |
| 55 | } |
| 56 | return errors.Join(syscall.Flock(int(l.file.Fd()), syscall.LOCK_UN), l.file.Close()) |
| 57 | } |
| 58 | |
| 59 | // Start acquires the process lock, opens ready tenants and starts catalog |
| 60 | // reconciliation. baseURL supplies their public address. ctx bounds startup; |
| 61 | // App.Close ends the running lifecycle. Calls on a started App return nil. |
| 62 | func (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 nil |
| 67 | } |
| 68 | a.mu.Lock() |
| 69 | closed := a.closed |
| 70 | 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 err |
| 77 | } |
| 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 = lock |
| 83 | a.lifecycle.started = true |
| 84 | a.lifecycle.baseURL = baseURL |
| 85 | runCtx, cancel := context.WithCancel(context.Background()) |
| 86 | a.lifecycle.cancel = cancel |
| 87 | done := make(chan struct{}) |
| 88 | a.lifecycle.done = done |
| 89 | 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 | return |
| 97 | 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 nil |
| 105 | } |
| 106 | |
| 107 | func (a *App) stopLifecycle() error { |
| 108 | a.lifecycle.mu.Lock() |
| 109 | if !a.lifecycle.started { |
| 110 | a.lifecycle.mu.Unlock() |
| 111 | return nil |
| 112 | } |
| 113 | a.lifecycle.started = false |
| 114 | done := a.lifecycle.done |
| 115 | a.lifecycle.cancel() |
| 116 | a.lifecycle.mu.Unlock() |
| 117 | <-done |
| 118 | return nil |
| 119 | } |
| 120 | |
| 121 | func (a *App) releaseProcessLock() error { |
| 122 | a.lifecycle.mu.Lock() |
| 123 | defer a.lifecycle.mu.Unlock() |
| 124 | lock := a.lifecycle.processLock |
| 125 | a.lifecycle.processLock = nil |
| 126 | return lock.Close() |
| 127 | } |
| 128 | |
| 129 | func (a *App) startCreatedTenant(ctx context.Context, meta catalog.Tenant) error { |
| 130 | a.lifecycle.mu.Lock() |
| 131 | base, started := a.lifecycle.baseURL, a.lifecycle.started |
| 132 | a.lifecycle.mu.Unlock() |
| 133 | if !started || meta.Status != catalog.StatusReady { |
| 134 | return nil |
| 135 | } |
| 136 | if meta.Name != a.cfg.DefaultTenant { |
| 137 | base += "/r/" + meta.Name |
| 138 | } |
| 139 | _, err := a.tenant(ctx, meta, base) |
| 140 | return err |
| 141 | } |
| 142 | |
| 143 | func (a *App) reconcileTenants(ctx context.Context, base string) error { |
| 144 | metas, err := a.catalog.List(ctx) |
| 145 | if err != nil { |
| 146 | return err |
| 147 | } |
| 148 | ready := make(map[string]bool, len(metas)) |
| 149 | for _, meta := range metas { |
| 150 | ready[meta.ID] = meta.Status == catalog.StatusReady |
| 151 | } |
| 152 | var stopped []*Tenant |
| 153 | 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 []error |
| 162 | for _, tenant := range stopped { |
| 163 | errs = append(errs, tenant.Close(ctx)) |
| 164 | } |
| 165 | for _, meta := range metas { |
| 166 | if !ready[meta.ID] { |
| 167 | continue |
| 168 | } |
| 169 | u := base |
| 170 | if meta.Name != a.cfg.DefaultTenant { |
| 171 | u += "/r/" + meta.Name |
| 172 | } |
| 173 | if _, err := a.tenant(ctx, meta, u); err != nil { |
| 174 | errs = append(errs, err) |
| 175 | } |
| 176 | } |
| 177 | return errors.Join(errs...) |
| 178 | } |