aboutsummaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
authorNikolay Govorov <me@govorov.online>2026-04-01 16:00:21 +0100
committerNikolay Govorov <me@govorov.online>2026-04-01 16:00:44 +0100
commit3a603f7e68365899564ac54ccc6f4887c948f15e (patch)
treef91dfdf1af36cc52dd08695cbe17fde921e9aefe
parentec0d5a51cba7cf14e30349545bbbe04868b81634 (diff)
downloadtar
tar.gz
tar.bz2
tar.lz
tar.xz
tar.zst
zip
Move execution to worker
Diffstat
-rw-r--r--cmd/mirumd/main.go158+134 −24
-rw-r--r--cmd/mirumw/client.go27+27 −0
-rw-r--r--cmd/mirumw/main.go5+3 −2
-rw-r--r--internal/protocol/backoff.go5+2 −3
-rw-r--r--internal/supervisor/supervisor.go6+4 −2
-rw-r--r--internal/supervisor/systemd.go5+4 −1
-rw-r--r--proto/mirum.proto31+31 −0
7 files changed, 205 insertions, 32 deletions
diff --git a/cmd/mirumd/main.go b/cmd/mirumd/main.go
index a959661..da78050 100644
--- a/cmd/mirumd/main.go
+++ b/cmd/mirumd/main.go
@@ -13,9 +13,9 @@ import (
"net"
"net/http"
"os"
+ "sync"
"time"
- "mrdimidium/mirum/internal/executor"
"mrdimidium/mirum/internal/forges"
"mrdimidium/mirum/internal/protocol"
"mrdimidium/mirum/internal/protocol/pb"
@@ -23,6 +23,10 @@ import (
"github.com/coreos/go-systemd/v22/activation"
"google.golang.org/grpc"
+ "google.golang.org/grpc/codes"
+ "google.golang.org/grpc/peer"
+ "google.golang.org/grpc/stats"
+ "google.golang.org/grpc/status"
"google.golang.org/protobuf/types/known/timestamppb"
"gopkg.in/yaml.v3"
)
@@ -41,24 +45,16 @@ var cfg = config{
var configFile = flag.String("config", "", "path to config file")
-func processPush(forge forges.Forge, ev *forges.PushEvent) {
- ctx := context.Background()
- log := slog.With("repo", ev.Owner+"/"+ev.Repo, "sha", ev.SHA[:8])
-
- if err := forge.SetStatus(ctx, ev, forges.StatusPending, "Build started"); err != nil {
- log.Error("set pending status", "err", err)
- }
-
- cloneURL := forge.AuthURL(ev.CloneURL)
+type taskMeta struct {
+ forge forges.Forge
+ event *forges.PushEvent
+}
- if err := executor.Run(cloneURL, ev.Branch); err != nil {
- log.Error("build failed", "err", err)
- _ = forge.SetStatus(ctx, ev, forges.StatusFailure, "Build failed")
- return
- }
+var taskCounter int64
- log.Info("build passed")
- _ = forge.SetStatus(ctx, ev, forges.StatusSuccess, "Build passed")
+func nextTaskID() string {
+ taskCounter++
+ return fmt.Sprintf("task-%d", taskCounter)
}
func main() {
@@ -83,6 +79,11 @@ func main() {
forge := &forges.GitHub{Secret: cfg.Secret, Token: cfg.Token}
+ srv := &mirumServer{
+ secret: []byte(cfg.Secret),
+ queue: make(chan *pb.Task, 100),
+ }
+
mux := http.NewServeMux()
mux.HandleFunc("GET /", func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "text/html; charset=utf-8")
@@ -109,10 +110,21 @@ func main() {
return
}
- slog.Info("push", "repo", ev.Owner+"/"+ev.Repo, "branch", ev.Branch, "sha", ev.SHA[:8])
- w.WriteHeader(http.StatusAccepted)
+ id := nextTaskID()
+ slog.Info("push", "repo", ev.Owner+"/"+ev.Repo, "branch", ev.Branch, "sha", ev.SHA[:8], "task", id)
+
+ srv.tasks.Store(id, taskMeta{forge: forge, event: ev})
+ _ = forge.SetStatus(context.Background(), ev, forges.StatusPending, "Queued")
- go processPush(forge, ev)
+ srv.queue <- &pb.Task{
+ Id: id,
+ CloneUrl: forge.AuthURL(ev.CloneURL),
+ Branch: ev.Branch,
+ Sha: ev.SHA,
+ RepoFullName: ev.Owner + "/" + ev.Repo,
+ }
+
+ w.WriteHeader(http.StatusAccepted)
})
grpcLn, httpLn, err := listeners()
@@ -121,8 +133,12 @@ func main() {
os.Exit(1)
}
- grpcSrv := grpc.NewServer(grpc.StreamInterceptor(streamTimeoutInterceptor))
- pb.RegisterMirumServer(grpcSrv, &mirumServer{secret: []byte(cfg.Secret)})
+ grpcSrv := grpc.NewServer(
+ grpc.UnaryInterceptor(srv.unaryInterceptor),
+ grpc.StreamInterceptor(srv.streamInterceptor),
+ grpc.StatsHandler(&connTracker{server: srv}),
+ )
+ pb.RegisterMirumServer(grpcSrv, srv)
httpSrv := &http.Server{Handler: mux}
slog.Info("listening", "grpc", grpcLn.Addr(), "http", httpLn.Addr())
@@ -155,7 +171,45 @@ func main() {
type mirumServer struct {
pb.UnimplementedMirumServer
- secret []byte
+ secret []byte
+ queue chan *pb.Task
+ tasks sync.Map // task_id → taskMeta
+ authedPeers sync.Map // peer addr string → true
+}
+
+func (s *mirumServer) Poll(ctx context.Context, req *pb.PollRequest) (*pb.Task, error) {
+ select {
+ case task := <-s.queue:
+ slog.Info("task dispatched", "id", task.Id, "repo", task.RepoFullName)
+ return task, nil
+ case <-ctx.Done():
+ return nil, ctx.Err()
+ }
+}
+
+func (s *mirumServer) Complete(ctx context.Context, result *pb.TaskResult) (*pb.CompleteResponse, error) {
+ meta, ok := s.tasks.LoadAndDelete(result.TaskId)
+ if !ok {
+ return nil, fmt.Errorf("unknown task: %s", result.TaskId)
+ }
+ m := meta.(taskMeta)
+
+ status := forges.StatusSuccess
+ desc := "Build passed"
+ if !result.Success {
+ status = forges.StatusFailure
+ desc = "Build failed"
+ if result.Error != "" {
+ desc = result.Error
+ }
+ }
+
+ if err := m.forge.SetStatus(ctx, m.event, status, desc); err != nil {
+ slog.Error("set status", "task", result.TaskId, "err", err)
+ }
+
+ slog.Info("task complete", "id", result.TaskId, "success", result.Success)
+ return &pb.CompleteResponse{}, nil
}
func (s *mirumServer) Handshake(stream pb.Mirum_HandshakeServer) error {
@@ -225,6 +279,9 @@ func (s *mirumServer) Handshake(stream pb.Mirum_HandshakeServer) error {
)
// Step 4: accept
+ if p, ok := peer.FromContext(stream.Context()); ok {
+ s.authedPeers.Store(p.Addr.String(), true)
+ }
return s.sendResult(stream, nil)
}
@@ -247,7 +304,34 @@ func (s *mirumServer) reject(stream pb.Mirum_HandshakeServer, reason string) err
return fmt.Errorf("%s", reason)
}
-func streamTimeoutInterceptor(srv any, ss grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
+func peerAddr(ctx context.Context) string {
+ if p, ok := peer.FromContext(ctx); ok {
+ return p.Addr.String()
+ }
+ return ""
+}
+
+func (s *mirumServer) requireAuth(ctx context.Context, method string) error {
+ if method == pb.Mirum_Handshake_FullMethodName {
+ return nil
+ }
+ if _, ok := s.authedPeers.Load(peerAddr(ctx)); !ok {
+ return status.Error(codes.Unauthenticated, "handshake required")
+ }
+ return nil
+}
+
+func (s *mirumServer) unaryInterceptor(ctx context.Context, req any, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (any, error) {
+ if err := s.requireAuth(ctx, info.FullMethod); err != nil {
+ return nil, err
+ }
+ return handler(ctx, req)
+}
+
+func (s *mirumServer) streamInterceptor(srv any, ss grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
+ if err := s.requireAuth(ss.Context(), info.FullMethod); err != nil {
+ return err
+ }
if info.FullMethod == pb.Mirum_Handshake_FullMethodName {
done := make(chan error, 1)
go func() { done <- handler(srv, ss) }()
@@ -261,6 +345,32 @@ func streamTimeoutInterceptor(srv any, ss grpc.ServerStream, info *grpc.StreamSe
return handler(srv, ss)
}
+// connTracker implements stats.Handler to clean up authedPeers on disconnect.
+type connTracker struct {
+ stats.Handler
+ server *mirumServer
+}
+
+func (t *connTracker) TagConn(ctx context.Context, info *stats.ConnTagInfo) context.Context {
+ return ctx
+}
+
+func (t *connTracker) TagRPC(ctx context.Context, info *stats.RPCTagInfo) context.Context {
+ return ctx
+}
+
+func (t *connTracker) HandleRPC(ctx context.Context, s stats.RPCStats) {}
+
+func (t *connTracker) HandleConn(ctx context.Context, s stats.ConnStats) {
+ if _, ok := s.(*stats.ConnEnd); ok {
+ addr := peerAddr(ctx)
+ if addr != "" {
+ t.server.authedPeers.Delete(addr)
+ slog.Debug("peer disconnected", "addr", addr)
+ }
+ }
+}
+
// listeners returns gRPC and HTTP listeners.
// With systemd socket activation it expects two named fds: "grpc" and "http".
// Without socket activation it falls back to cfg.GrpcAddr and cfg.WwwAddr.
diff --git a/cmd/mirumw/client.go b/cmd/mirumw/client.go
index 779ab11..57ca17b 100644
--- a/cmd/mirumw/client.go
+++ b/cmd/mirumw/client.go
@@ -12,6 +12,7 @@ import (
"os"
"time"
+ "mrdimidium/mirum/internal/executor"
"mrdimidium/mirum/internal/protocol"
"mrdimidium/mirum/internal/protocol/pb"
@@ -70,6 +71,32 @@ func (c *client) close() error {
return c.conn.Close()
}
+func (c *client) work(ctx context.Context) error {
+ for ctx.Err() == nil {
+ task, err := c.handle.Poll(ctx, &pb.PollRequest{})
+ if err != nil {
+ return fmt.Errorf("poll: %w", err)
+ }
+
+ slog.Info("task received", "id", task.Id, "repo", task.RepoFullName)
+
+ execErr := executor.Run(task.CloneUrl, task.Branch)
+
+ result := &pb.TaskResult{TaskId: task.Id, Success: execErr == nil}
+ if execErr != nil {
+ result.Error = execErr.Error()
+ slog.Error("task failed", "id", task.Id, "err", execErr)
+ } else {
+ slog.Info("task passed", "id", task.Id)
+ }
+
+ if _, err := c.handle.Complete(ctx, result); err != nil {
+ return fmt.Errorf("complete: %w", err)
+ }
+ }
+ return ctx.Err()
+}
+
func (c *client) handshake(ctx context.Context) error {
stream, err := c.handle.Handshake(ctx)
if err != nil {
diff --git a/cmd/mirumw/main.go b/cmd/mirumw/main.go
index 0980a9a..1a59e8b 100644
--- a/cmd/mirumw/main.go
+++ b/cmd/mirumw/main.go
@@ -45,8 +45,9 @@ func main() {
slog.Info("connected", "server", cfg.Server)
backoff.Reset()
- // TODO: c.Work(ctx) — poll tasks, execute, report
- <-ctx.Done()
+ if err := c.work(ctx); err != nil && ctx.Err() == nil {
+ slog.Error("work loop failed", "err", err)
+ }
c.close()
}
diff --git a/internal/protocol/backoff.go b/internal/protocol/backoff.go
index 8f07892..ca5e193 100644
--- a/internal/protocol/backoff.go
+++ b/internal/protocol/backoff.go
@@ -10,9 +10,8 @@ import (
)
type Backoff struct {
- attempt int
- Min time.Duration
- Max time.Duration
+ attempt int
+ Min, Max time.Duration
}
func NewBackoff() *Backoff {
diff --git a/internal/supervisor/supervisor.go b/internal/supervisor/supervisor.go
index f6fb619..6005dd5 100644
--- a/internal/supervisor/supervisor.go
+++ b/internal/supervisor/supervisor.go
@@ -41,8 +41,10 @@ func Detect() Supervisor {
type noop struct{}
-func (*noop) Ready() {}
-func (*noop) Stopping() {}
+func (*noop) Ready() {}
+
+func (*noop) Stopping() {}
+
func (*noop) StartWatchdog() {}
func (*noop) WaitForStop(ctx context.Context) context.Context {
diff --git a/internal/supervisor/systemd.go b/internal/supervisor/systemd.go
index f17307c..c4d6ff4 100644
--- a/internal/supervisor/systemd.go
+++ b/internal/supervisor/systemd.go
@@ -23,7 +23,7 @@ func detectSystemd() bool {
func (*systemd) WaitForStop(ctx context.Context) context.Context {
ctx, stop := signal.NotifyContext(ctx, syscall.SIGTERM, syscall.SIGINT)
- _ = stop
+ _ = stop // stop allows you to cancel the observer, but it's not particularly useful
return ctx
}
@@ -44,16 +44,19 @@ func (*systemd) StartWatchdog() {
if usecStr == "" {
return
}
+
usec, err := strconv.ParseInt(usecStr, 10, 64)
if err != nil || usec <= 0 {
return
}
+
interval := time.Duration(usec) * time.Microsecond / 2
for {
if _, err := daemon.SdNotify(false, daemon.SdNotifyWatchdog); err != nil {
slog.Warn("sd_notify watchdog", "err", err)
return
}
+
time.Sleep(interval)
}
}
diff --git a/proto/mirum.proto b/proto/mirum.proto
index c8768b8..80b0ea1 100644
--- a/proto/mirum.proto
+++ b/proto/mirum.proto
@@ -9,14 +9,25 @@ option go_package = "internal/protocol/pb";
import "google/protobuf/timestamp.proto";
+// Describes the contract between the worker and the server.
+// GRPC is the only contract between them, so to implement your own worker,
+// you only need to implement this service.
service Mirum {
// Handshake performs mutual authentication via HMAC-SHA256 challenge-response.
+ // The server will not accept any calls until a handshake is completed.
//
// Step 1 (W→S): WorkerChallenge — worker sends random nonce.
// Step 2 (S→W): ServerChallenge — server sends nonce + HMAC(secret, w_nonce || s_nonce).
// Step 3 (W→S): WorkerProof — worker sends HMAC(secret, s_nonce || w_nonce) + metadata.
// Step 4 (S→W): ServerResult — server accepts or rejects.
rpc Handshake(stream HandshakeIn) returns (stream HandshakeOut);
+
+ // When a worker has free resources, it requests a task from the server.
+ // The call will block if the server currently has no tasks.
+ rpc Poll(PollRequest) returns (Task);
+
+ // When the task is completed, the worker reports the result to the server.
+ rpc Complete(TaskResult) returns (CompleteResponse);
}
message Version {
@@ -168,3 +179,23 @@ message ServerResult {
Version server_version = 2;
google.protobuf.Timestamp server_time = 3;
}
+
+message PollRequest {
+
+}
+
+message Task {
+ string id = 1;
+ string clone_url = 2;
+ string branch = 3;
+ string sha = 4;
+ string repo_full_name = 5;
+}
+
+message TaskResult {
+ string task_id = 1;
+ bool success = 2;
+ string error = 3;
+}
+
+message CompleteResponse {}
generated by Gilti