refactor: fix all gofmt issues and further reduce daemon cyclomatic complexity

Author: alien43

cmd/conch/main.go

@@ -260,7 +260,6 @@ func runElectReadOnly(ctx context.Context, logger *slog.Logger, endpoints []stri
 	}
 }
 
-
 func handleSema(args []string) {
 	fs := flag.NewFlagSet("sema", flag.ExitOnError)
 

internal/cron/cron.go

@@ -595,49 +595,56 @@ func (cd *Conchd) Run(ctx context.Context) error {
 				return fmt.Errorf("watch channel closed")
 			}
 			for _, ev := range wresp.Events {
-				name := strings.TrimPrefix(string(ev.Kv.Key), core.CronJobPrefix())
-				if ev.Type == clientv3.EventTypeDelete {
-					if cancel, ok := cancels[name]; ok {
-						cancel()
-						delete(cancels, name)
-					}
-					delete(jobs, name)
-					cd.logger.Info("job removed", "name", name)
-				} else {
-					var spec JobSpec
-					if err := json.Unmarshal(ev.Kv.Value, &spec); err == nil {
-						oldSpec, exists := jobs[name]
-						if exists && oldSpec.Schedule == spec.Schedule && oldSpec.RunTTL == spec.RunTTL && len(oldSpec.Cmd) == len(spec.Cmd) {
-							cmdChanged := false
-							for i := range spec.Cmd {
-								if spec.Cmd[i] != oldSpec.Cmd[i] {
-									cmdChanged = true
-									break
-								}
-							}
-							if !cmdChanged {
-								// Execution spec is unchanged. Update jobs map for metadata, but do NOT restart scheduler.
-								jobs[name] = spec
-								cd.logger.Debug("job touched but execution spec unchanged; not restarting scheduler", "name", name)
-								continue
-							}
-						}
-
-						if cancel, ok := cancels[name]; ok {
-							cancel()
-						}
-						jobs[name] = spec
-						jobCtx, cancel := context.WithCancel(ctx)
-						cancels[name] = cancel
-						go runJobScheduler(ctx, jobCtx, cd.logger, cd.client, sess, name, spec)
-						cd.logger.Info("job added/updated", "name", name)
-					}
-				}
+				cd.handleWatchEvent(ctx, ev, jobs, cancels, sess)
 			}
 		}
 	}
 }
 
+func (cd *Conchd) handleWatchEvent(ctx context.Context, ev *clientv3.Event, jobs map[string]JobSpec, cancels map[string]context.CancelFunc, sess *core.CoreSession) {
+	name := strings.TrimPrefix(string(ev.Kv.Key), core.CronJobPrefix())
+	if ev.Type == clientv3.EventTypeDelete {
+		if cancel, ok := cancels[name]; ok {
+			cancel()
+			delete(cancels, name)
+		}
+		delete(jobs, name)
+		cd.logger.Info("job removed", "name", name)
+		return
+	}
+
+	var spec JobSpec
+	if err := json.Unmarshal(ev.Kv.Value, &spec); err != nil {
+		return
+	}
+
+	oldSpec, exists := jobs[name]
+	if exists && oldSpec.Schedule == spec.Schedule && oldSpec.RunTTL == spec.RunTTL && len(oldSpec.Cmd) == len(spec.Cmd) {
+		cmdChanged := false
+		for i := range spec.Cmd {
+			if spec.Cmd[i] != oldSpec.Cmd[i] {
+				cmdChanged = true
+				break
+			}
+		}
+		if !cmdChanged {
+			// Execution spec is unchanged. Update jobs map for metadata, but do NOT restart scheduler.
+			jobs[name] = spec
+			cd.logger.Debug("job touched but execution spec unchanged; not restarting scheduler", "name", name)
+			return
+		}
+	}
+
+	if cancel, ok := cancels[name]; ok {
+		cancel()
+	}
+	jobs[name] = spec
+	jobCtx, cancel := context.WithCancel(ctx)
+	cancels[name] = cancel
+	go runJobScheduler(ctx, jobCtx, cd.logger, cd.client, sess, name, spec)
+	cd.logger.Info("job added/updated", "name", name)
+}
+
 func runJobScheduler(daemonCtx, jobCtx context.Context, logger *slog.Logger, client *clientv3.Client, sess *core.CoreSession, name string, spec JobSpec) {
 	parser := cron.NewParser(cron.Minute | cron.Hour | cron.Dom | cron.Month | cron.Dow | cron.Descriptor)
 	sched, err := parser.Parse(spec.Schedule)

internal/elect/elect.go

@@ -346,4 +346,3 @@ func updateCandidates(candidates []candidate, events []*clientv3.Event) []candid
 	}
 	return candidates
 }
-

internal/sema/sema.go

@@ -106,56 +106,68 @@ func (sh *SemaHolder) acquireInternal(ctx context.Context) (int64, error) {
 		prefix := core.SemaPrefix(sh.name, sh.max)
 
 		for {
-			// 2. Fetch all keys under the prefix sorted by create-revision
-			getResp, err := sh.client.Get(ctx, prefix,
-				clientv3.WithPrefix(),
-				clientv3.WithSort(clientv3.SortByCreateRevision, clientv3.SortAscend),
-			)
+			rev, acquired, err := sh.checkRankAndWatch(ctx, prefix)
 			if err != nil {
 				return 0, err
 			}
-
-			// Find our rank
-			rank := -1
-			for i, kv := range getResp.Kvs {
-				if string(kv.Key) == sh.key {
-					rank = i
-					break
-				}
+			if acquired {
+				return rev, nil
 			}
+		}
+	}
+}
 
-			if rank == -1 {
-				return 0, fmt.Errorf("our key was deleted from etcd")
-			}
+func (sh *SemaHolder) checkRankAndWatch(ctx context.Context, prefix string) (int64, bool, error) {
+	// 2. Fetch all keys under the prefix sorted by create-revision
+	getResp, err := sh.client.Get(ctx, prefix,
+		clientv3.WithPrefix(),
+		clientv3.WithSort(clientv3.SortByCreateRevision, clientv3.SortAscend),
+	)
+	if err != nil {
+		return 0, false, err
+	}
 
-			// 3. If rank < N, we hold a slot!
-			if rank < sh.max {
-				sh.acquired = true
-				return getResp.Kvs[rank].CreateRevision, nil
-			}
+	// Find our rank
+	rank := -1
+	for i, kv := range getResp.Kvs {
+		if string(kv.Key) == sh.key {
+			rank = i
+			break
+		}
+	}
 
-			// Non-blocking means we don't wait if rank >= max
-			if sh.waitLimit == 0 {
-				return 0, context.DeadlineExceeded // Treat non-block as immediate timeout
-			}
+	if rank == -1 {
+		return 0, false, fmt.Errorf("our key was deleted from etcd")
+	}
 
-			// 4. Watch for deletion of the key at rank - max
-			targetKey := string(getResp.Kvs[rank-sh.max].Key)
+	// 3. If rank < N, we hold a slot!
+	if rank < sh.max {
+		sh.acquired = true
+		return getResp.Kvs[rank].CreateRevision, true, nil
+	}
 
-			// Watch from the revision of our Get response to avoid races
-			err = sh.watchKeyDeletion(ctx, targetKey, getResp.Header.Revision, "session lost while waiting for semaphore slot")
-			if err != nil {
-				return 0, err
-			}
+	// Non-blocking means we don't wait if rank >= max
+	if sh.waitLimit == 0 {
+		return 0, false, context.DeadlineExceeded // Treat non-block as immediate timeout
+	}
 
-			// Check context done
-			if ctx.Err() != nil {
-				return 0, ctx.Err()
-			}
-		}
+	// 4. Watch for deletion of the key at rank - max
+	targetKey := string(getResp.Kvs[rank-sh.max].Key)
+
+	// Watch from the revision of our Get response to avoid races
+	err = sh.watchKeyDeletion(ctx, targetKey, getResp.Header.Revision, "session lost while waiting for semaphore slot")
+	if err != nil {
+		return 0, false, err
 	}
+
+	// Check context done
+	if ctx.Err() != nil {
+		return 0, false, ctx.Err()
+	}
+	return 0, false, nil
 }
 
+
 func (sh *SemaHolder) watchKeyDeletion(ctx context.Context, targetKey string, startRev int64, sessionErrMsg string) error {
 	watchCtx, watchCancel := context.WithCancel(ctx)
 	defer watchCancel()