c4dde1c3f98a5f6d8a182e1ad2b460b24aeb71e7
- Author
- Ayman Bagabas <ayman.bagabas@gmail.com>
- Committer
- Ayman Bagabas <ayman.bagabas@gmail.com>
- Date
Message
feat,fix: add task manager Implement a task manager that can run different tasks given a unique ID. This is needed to accommodate expensive tasks like importing a large repository. The current behavior uses the connection's context (the SSH connection) to import the repository. However, if the server has defined an SSH `idle_timeout`, `max_timeout`, and/or the connection drops, Soft Serve cancels the git clone process and aborts importing the repository. Instead, we add the import task to the "task manager" and wait on the connection context. If a task already exists for the same repository, return `Error: import already in progress`. Fixes: https://github.com/charmbracelet/soft-serve/issues/348
Diff
This diff is truncated to protect this page.
1diff --git a/server/backend/backend.go b/server/backend/backend.go
2index 586d95132504a852dc8fa339de464d3360aca59d..ba9ad61d3713dd18f72a8a9e91d989ddcc370fe1 100644
3--- a/server/backend/backend.go
4+++ b/server/backend/backend.go
5@@ -7,17 +7,19 @@ import (
6 "github.com/charmbracelet/soft-serve/server/config"
7 "github.com/charmbracelet/soft-serve/server/db"
8 "github.com/charmbracelet/soft-serve/server/store"
9+ "github.com/charmbracelet/soft-serve/server/task"
10 )
11
12 // Backend is the Soft Serve backend that handles users, repositories, and
13 // server settings management and operations.
14 type Backend struct {
15- ctx context.Context
16- cfg *config.Config
17- db *db.DB
18- store store.Store
19- logger *log.Logger
20- cache *cache
21+ ctx context.Context
22+ cfg *config.Config
23+ db *db.DB
24+ store store.Store
25+ logger *log.Logger
26+ cache *cache
27+ manager *task.Manager
28 }
29
30 // New returns a new Soft Serve backend.
31@@ -25,11 +27,12 @@ func New(ctx context.Context, cfg *config.Config, db *db.DB) *Backend {
32 dbstore := store.FromContext(ctx)
33 logger := log.FromContext(ctx).WithPrefix("backend")
34 b := &Backend{
35- ctx: ctx,
36- cfg: cfg,
37- db: db,
38- store: dbstore,
39- logger: logger,
40+ ctx: ctx,
41+ cfg: cfg,
42+ db: db,
43+ store: dbstore,
44+ logger: logger,
45+ manager: task.NewManager(ctx),
46 }
47
48 // TODO: implement a proper caching interface
49diff --git a/server/backend/repo.go b/server/backend/repo.go
50index c9c6a7257e2674909cf7d8e44811a956d3562129..2d8633cc3961c6883cea0942eef2aea67cadaa78 100644
51--- a/server/backend/repo.go
52+++ b/server/backend/repo.go
53@@ -19,6 +19,7 @@ import (
54 "github.com/charmbracelet/soft-serve/server/lfs"
55 "github.com/charmbracelet/soft-serve/server/proto"
56 "github.com/charmbracelet/soft-serve/server/storage"
57+ "github.com/charmbracelet/soft-serve/server/task"
58 "github.com/charmbracelet/soft-serve/server/utils"
59 )
60
61@@ -91,7 +92,8 @@ func (d *Backend) CreateRepository(ctx context.Context, name string, user proto.
62 }
63
64 // ImportRepository imports a repository from remote.
65-func (d *Backend) ImportRepository(ctx context.Context, name string, user proto.User, remote string, opts proto.RepositoryOptions) (proto.Repository, error) {
66+// XXX: This a expensive operation and should be run in a goroutine.
67+func (d *Backend) ImportRepository(_ context.Context, name string, user proto.User, remote string, opts proto.RepositoryOptions) (proto.Repository, error) {
68 name = utils.SanitizeRepo(name)
69 if err := utils.ValidateRepo(name); err != nil {
70 return nil, err
71@@ -100,91 +102,110 @@ func (d *Backend) ImportRepository(ctx context.Context, name string, user proto.
72 repo := name + ".git"
73 rp := filepath.Join(d.reposPath(), repo)
74
75+ tid := "import:" + name
76+ if d.manager.Exists(tid) {
77+ return nil, task.ErrAlreadyStarted
78+ }
79+
80 if _, err := os.Stat(rp); err == nil || os.IsExist(err) {
81 return nil, proto.ErrRepoExist
82 }
83
84- copts := git.CloneOptions{
85- Bare: true,
86- Mirror: opts.Mirror,
87- Quiet: true,
88- CommandOptions: git.CommandOptions{
89- Timeout: -1,
90- Context: ctx,
91- Envs: []string{
92- fmt.Sprintf(`GIT_SSH_COMMAND=ssh -o UserKnownHostsFile="%s" -o StrictHostKeyChecking=no -i "%s"`,
93- filepath.Join(d.cfg.DataPath, "ssh", "known_hosts"),
94- d.cfg.SSH.ClientKeyPath,
95- ),
96+ done := make(chan error, 1)
97+ repoc := make(chan proto.Repository, 1)
98+ d.logger.Info("importing repository", "name", name, "remote", remote, "path", rp)
99+ d.manager.Add(tid, func(ctx context.Context) (err error) {
100+ copts := git.CloneOptions{
101+ Bare: true,
102+ Mirror: opts.Mirror,
103+ Quiet: true,
104+ CommandOptions: git.CommandOptions{
105+ Timeout: -1,
106+ Context: ctx,
107+ Envs: []string{
108+ fmt.Sprintf(`GIT_SSH_COMMAND=ssh -o UserKnownHostsFile="%s" -o StrictHostKeyChecking=no -i "%s"`,
109+ filepath.Join(d.cfg.DataPath, "ssh", "known_hosts"),
110+ d.cfg.SSH.ClientKeyPath,
111+ ),
112+ },
113 },
114- },
115- }
116-
117- if err := git.Clone(remote, rp, copts); err != nil {
118- d.logger.Error("failed to clone repository", "err", err, "mirror", opts.Mirror, "remote", remote, "path", rp)
119- // Cleanup the mess!
120- if rerr := os.RemoveAll(rp); rerr != nil {
121- err = errors.Join(err, rerr)
122 }
123
124- return nil, err
125- }
126+ if err := git.Clone(remote, rp, copts); err != nil {
127+ d.logger.Error("failed to clone repository", "err", err, "mirror", opts.Mirror, "remote", remote, "path", rp)
128+ // Cleanup the mess!
129+ if rerr := os.RemoveAll(rp); rerr != nil {
130+ err = errors.Join(err, rerr)
131+ }
132
133- r, err := d.CreateRepository(ctx, name, user, opts)
134- if err != nil {
135- d.logger.Error("failed to create repository", "err", err, "name", name)
136- return nil, err
137- }
138+ return err
139+ }
140
141- defer func() {
142+ r, err := d.CreateRepository(ctx, name, user, opts)
143 if err != nil {
144- if rerr := d.DeleteRepository(ctx, name); rerr != nil {
145- d.logger.Error("failed to delete repository", "err", rerr, "name", name)
146+ d.logger.Error("failed to create repository", "err", err, "name", name)
147+ return err
148+ }
149+
150+ defer func() {
151+ if err != nil {
152+ if rerr := d.DeleteRepository(ctx, name); rerr != nil {
153diff --git a/server/ssh/cmd/import.go b/server/ssh/cmd/import.go
154index b34b46f613389ed41a779060549e2ac760d8c4b9..85cb2fb8f3e09a464a8dc9efccaa5d3defe4ebfb 100644
155--- a/server/ssh/cmd/import.go
156+++ b/server/ssh/cmd/import.go
157@@ -1,8 +1,11 @@
158 package cmd
159
160 import (
161+ "errors"
162+
163 "github.com/charmbracelet/soft-serve/server/backend"
164 "github.com/charmbracelet/soft-serve/server/proto"
165+ "github.com/charmbracelet/soft-serve/server/task"
166 "github.com/spf13/cobra"
167 )
168
169@@ -36,8 +39,13 @@ func importCommand() *cobra.Command {
170 LFS: lfs,
171 LFSEndpoint: lfsEndpoint,
172 }); err != nil {
173+ if errors.Is(err, task.ErrAlreadyStarted) {
174+ return errors.New("import already in progress")
175+ }
176+
177 return err
178 }
179+
180 return nil
181 },
182 }
183diff --git a/server/task/manager.go b/server/task/manager.go
184new file mode 100644
185index 0000000000000000000000000000000000000000..4f8763711d9478820774c04a04a44c30e3c5d7b2
186--- /dev/null
187+++ b/server/task/manager.go
188@@ -0,0 +1,116 @@
189+package task
190+
191+import (
192+ "context"
193+ "errors"
194+ "sync"
195+ "sync/atomic"
196+)
197+
198+var (
199+ // ErrNotFound is returned when a process is not found.
200+ ErrNotFound = errors.New("task not found")
201+
202+ // ErrAlreadyStarted is returned when a process is already started.
203+ ErrAlreadyStarted = errors.New("task already started")
204+)
205+
206+// Task is a task that can be started and stopped.
207+type Task struct {
208+ id string
209+ fn func(context.Context) error
210+ started atomic.Bool
211+ ctx context.Context
212+ cancel context.CancelFunc
213+ err error
214+}
215+
216+// Manager manages tasks.
217+type Manager struct {
218+ m sync.Map
219+ ctx context.Context
220+}
221+
222+// NewManager returns a new task manager.
223+func NewManager(ctx context.Context) *Manager {
224+ return &Manager{
225+ m: sync.Map{},
226+ ctx: ctx,
227+ }
228+}
229+
230+// Add adds a task to the manager.
231+// If the process already exists, it is a no-op.
232+func (m *Manager) Add(id string, fn func(context.Context) error) {
233+ if m.Exists(id) {
234+ return
235+ }
236+
237+ ctx, cancel := context.WithCancel(m.ctx)
238+ m.m.Store(id, &Task{
239+ id: id,
240+ fn: fn,
241+ ctx: ctx,
242+ cancel: cancel,
243+ })
244+}
245+
246+// Stop stops the task and removes it from the manager.
247+func (m *Manager) Stop(id string) error {
248+ v, ok := m.m.Load(id)
249+ if !ok {
250+ return ErrNotFound
251+ }
252+
253+ p := v.(*Task)
254+ p.cancel()
255+
256+ m.m.Delete(id)
257+ return nil
258+}
259+
260+// Exists checks if a task exists.
261+func (m *Manager) Exists(id string) bool {
262+ _, ok := m.m.Load(id)
263+ return ok
264+}
265+
266+// Run starts the task if it exists.
267+// Otherwise, it waits for the process to finish.
268+func (m *Manager) Run(id string, done chan<- error) {
269+ v, ok := m.m.Load(id)
270+ if !ok {
271+ done <- ErrNotFound
272+ return
273+ }
274+
275+ p := v.(*Task)
276+ if p.started.Load() {
277+ <-p.ctx.Done()
278+ if p.err != nil {
279+ done <- p.err
280+ return
281+ }
282+
283+ done <- p.ctx.Err()
284+ }
285+
286+ p.started.Store(true)
287+ m.m.Store(id, p)