diff options
Diffstat (limited to 'internal/transport')
| -rw-r--r-- | internal/transport/deliver.go | 19 | ||||
| -rw-r--r-- | internal/transport/delivery/delivery.go | 331 | ||||
| -rw-r--r-- | internal/transport/delivery/delivery_test.go | 265 | ||||
| -rw-r--r-- | internal/transport/delivery/worker.go | 298 | ||||
| -rw-r--r-- | internal/transport/delivery/worker_test.go | 220 |
5 files changed, 705 insertions, 428 deletions
diff --git a/internal/transport/deliver.go b/internal/transport/deliver.go index a7e73465d..30435b86f 100644 --- a/internal/transport/deliver.go +++ b/internal/transport/deliver.go @@ -18,12 +18,12 @@ package transport import ( + "bytes" "context" "encoding/json" "net/http" "net/url" - "codeberg.org/gruf/go-byteutil" apiutil "github.com/superseriousbusiness/gotosocial/internal/api/util" "github.com/superseriousbusiness/gotosocial/internal/config" "github.com/superseriousbusiness/gotosocial/internal/gtscontext" @@ -130,25 +130,28 @@ func (t *transport) prepare( *delivery.Delivery, error, ) { - url := to.String() - - // Use rewindable reader for body. - var body byteutil.ReadNopCloser - body.Reset(data) - // Prepare POST signer. sign := t.signPOST(data) + // Use *bytes.Reader for request body, + // as NewRequest() automatically will + // set .GetBody and content-length. + // (this handles necessary rewinding). + body := bytes.NewReader(data) + // Update to-be-used request context with signing details. ctx = gtscontext.SetOutgoingPublicKeyID(ctx, t.pubKeyID) ctx = gtscontext.SetHTTPClientSignFunc(ctx, sign) // Prepare a new request with data body directed at URL. - r, err := http.NewRequestWithContext(ctx, "POST", url, &body) + r, err := http.NewRequestWithContext(ctx, "POST", to.String(), body) if err != nil { return nil, gtserror.Newf("error preparing request: %w", err) } + // Set our predefined controller user-agent. + r.Header.Set("User-Agent", t.controller.userAgent) + // Set the standard ActivityPub content-type + charset headers. r.Header.Add("Content-Type", string(apiutil.AppActivityLDJSON)) r.Header.Add("Accept-Charset", "utf-8") diff --git a/internal/transport/delivery/delivery.go b/internal/transport/delivery/delivery.go index 1e9126b2e..1e3ebb054 100644 --- a/internal/transport/delivery/delivery.go +++ b/internal/transport/delivery/delivery.go @@ -18,16 +18,13 @@ package delivery import ( - "context" - "slices" + "bytes" + "encoding/json" + "io" + "net/http" "time" - "codeberg.org/gruf/go-runners" - "codeberg.org/gruf/go-structr" - "github.com/superseriousbusiness/gotosocial/internal/gtscontext" "github.com/superseriousbusiness/gotosocial/internal/httpclient" - "github.com/superseriousbusiness/gotosocial/internal/queue" - "github.com/superseriousbusiness/gotosocial/internal/util" ) // Delivery wraps an httpclient.Request{} @@ -36,6 +33,9 @@ import ( // be indexed (and so, dropped from queue) // by any of these possible ID IRIs. type Delivery struct { + // PubKeyID is the signing public key + // ID of the actor performing request. + PubKeyID string // ActorID contains the ActivityPub // actor ID IRI (if any) of the activity @@ -61,273 +61,98 @@ type Delivery struct { next time.Time } -func (dlv *Delivery) backoff() time.Duration { - if dlv.next.IsZero() { - return 0 - } - return time.Until(dlv.next) +// delivery is an internal type +// for Delivery{} that provides +// a json serialize / deserialize +// able shape that minimizes data. +type delivery struct { + PubKeyID string `json:"pub_key_id,omitempty"` + ActorID string `json:"actor_id,omitempty"` + ObjectID string `json:"object_id,omitempty"` + TargetID string `json:"target_id,omitempty"` + Method string `json:"method,omitempty"` + Header map[string][]string `json:"header,omitempty"` + URL string `json:"url,omitempty"` + Body []byte `json:"body,omitempty"` } -// WorkerPool wraps multiple Worker{}s in -// a singular struct for easy multi start/stop. -type WorkerPool struct { +// Serialize will serialize the delivery data as data blob for storage, +// note that this will flatten some of the data, dropping signing funcs. +func (dlv *Delivery) Serialize() ([]byte, error) { + var body []byte - // Client defines httpclient.Client{} - // passed to each of delivery pool Worker{}s. - Client *httpclient.Client + if dlv.Request.GetBody != nil { + // Fetch a fresh copy of request body. + rbody, err := dlv.Request.GetBody() + if err != nil { + return nil, err + } - // Queue is the embedded queue.StructQueue{} - // passed to each of delivery pool Worker{}s. - Queue queue.StructQueue[*Delivery] + // Read request body into memory. + body, err = io.ReadAll(rbody) - // internal fields. - workers []*Worker -} + // Done with body. + _ = rbody.Close() -// Init will initialize the Worker{} pool -// with given http client, request queue to pull -// from and number of delivery workers to spawn. -func (p *WorkerPool) Init(client *httpclient.Client) { - p.Client = client - p.Queue.Init(structr.QueueConfig[*Delivery]{ - Indices: []structr.IndexConfig{ - {Fields: "ActorID", Multiple: true}, - {Fields: "ObjectID", Multiple: true}, - {Fields: "TargetID", Multiple: true}, - }, - }) -} - -// Start will attempt to start 'n' Worker{}s. -func (p *WorkerPool) Start(n int) { - // Check whether workers are - // set (is already running). - ok := (len(p.workers) > 0) - if ok { - return + if err != nil { + return nil, err + } } - // Allocate new workers slice. - p.workers = make([]*Worker, n) - for i := range p.workers { - - // Allocate new Worker{}. - p.workers[i] = new(Worker) - p.workers[i].Client = p.Client - p.workers[i].Queue = &p.Queue - - // Attempt to start worker. - // Return bool not useful - // here, as true = started, - // false = already running. - _ = p.workers[i].Start() - } + // Marshal as internal JSON type. + return json.Marshal(delivery{ + PubKeyID: dlv.PubKeyID, + ActorID: dlv.ActorID, + ObjectID: dlv.ObjectID, + TargetID: dlv.TargetID, + Method: dlv.Request.Method, + Header: dlv.Request.Header, + URL: dlv.Request.URL.String(), + Body: body, + }) } -// Stop will attempt to stop contained Worker{}s. -func (p *WorkerPool) Stop() { - // Check whether workers are - // set (is currently running). - ok := (len(p.workers) == 0) - if ok { - return - } - - // Stop all running workers. - for i := range p.workers { +// Deserialize will attempt to deserialize a blob of task data, +// which will involve unflattening previously serialized data and +// leave delivery incomplete, still requiring signing func setup. +func (dlv *Delivery) Deserialize(data []byte) error { + var idlv delivery - // return bool not useful - // here, as true = stopped, - // false = never running. - _ = p.workers[i].Stop() + // Unmarshal as internal JSON type. + err := json.Unmarshal(data, &idlv) + if err != nil { + return err } - // Unset workers slice. - p.workers = p.workers[:0] -} - -// Worker wraps an httpclient.Client{} to feed -// from queue.StructQueue{} for ActivityPub reqs -// to deliver. It does so while prioritizing new -// queued requests over backlogged retries. -type Worker struct { - - // Client is the httpclient.Client{} that - // delivery worker will use for requests. - Client *httpclient.Client - - // Queue is the Delivery{} message queue - // that delivery worker will feed from. - Queue *queue.StructQueue[*Delivery] + // Copy over simplest fields. + dlv.PubKeyID = idlv.PubKeyID + dlv.ActorID = idlv.ActorID + dlv.ObjectID = idlv.ObjectID + dlv.TargetID = idlv.TargetID - // internal fields. - backlog []*Delivery - service runners.Service -} - -// Start will attempt to start the Worker{}. -func (w *Worker) Start() bool { - return w.service.GoRun(w.run) -} + var body io.Reader -// Stop will attempt to stop the Worker{}. -func (w *Worker) Stop() bool { - return w.service.Stop() -} - -// run wraps process to restart on any panic. -func (w *Worker) run(ctx context.Context) { - if w.Client == nil || w.Queue == nil { - panic("not yet initialized") - } - util.Must(func() { w.process(ctx) }) -} - -// process is the main delivery worker processing routine. -func (w *Worker) process(ctx context.Context) bool { - if w.Client == nil || w.Queue == nil { - // we perform this check here just - // to ensure the compiler knows these - // variables aren't nil in the loop, - // even if already checked by caller. - panic("not yet initialized") + if idlv.Body != nil { + // Create new body reader from data. + body = bytes.NewReader(idlv.Body) } -loop: - for { - // Get next delivery. - dlv, ok := w.next(ctx) - if !ok { - return true - } - - // Check whether backoff required. - const min = 100 * time.Millisecond - if d := dlv.backoff(); d > min { - - // Start backoff sleep timer. - backoff := time.NewTimer(d) - - select { - case <-ctx.Done(): - // Main ctx - // cancelled. - backoff.Stop() - return true - - case <-w.Queue.Wait(): - // A new message was - // queued, re-add this - // to backlog + retry. - w.pushBacklog(dlv) - backoff.Stop() - continue loop - - case <-backoff.C: - // success! - } - } - - // Attempt delivery of AP request. - rsp, retry, err := w.Client.DoOnce( - &dlv.Request, - ) - - if err == nil { - // Ensure body closed. - _ = rsp.Body.Close() - continue loop - } - - if !retry { - // Drop deliveries when no - // retry requested, or they - // reached max (either). - continue loop - } - - // Determine next delivery attempt. - backoff := dlv.Request.BackOff() - dlv.next = time.Now().Add(backoff) - - // Push to backlog. - w.pushBacklog(dlv) + // Create a new request object from unmarshaled details. + r, err := http.NewRequest(idlv.Method, idlv.URL, body) + if err != nil { + return err } -} - -// next gets the next available delivery, blocking until available if necessary. -func (w *Worker) next(ctx context.Context) (*Delivery, bool) { -loop: - for { - // Try pop next queued. - dlv, ok := w.Queue.Pop() - - if !ok { - // Check the backlog. - if len(w.backlog) > 0 { - - // Sort by 'next' time. - sortDeliveries(w.backlog) - - // Pop next delivery. - dlv := w.popBacklog() - - return dlv, true - } - select { - // Backlog is empty, we MUST - // block until next enqueued. - case <-w.Queue.Wait(): - continue loop + // Wrap request in httpclient type. + dlv.Request = httpclient.WrapRequest(r) - // Worker was stopped. - case <-ctx.Done(): - return nil, false - } - } - - // Replace request context for worker state canceling. - ctx := gtscontext.WithValues(ctx, dlv.Request.Context()) - dlv.Request.Request = dlv.Request.Request.WithContext(ctx) - - return dlv, true - } + return nil } -// popBacklog pops next available from the backlog. -func (w *Worker) popBacklog() *Delivery { - if len(w.backlog) == 0 { - return nil +// backoff returns a valid (>= 0) backoff duration. +func (dlv *Delivery) backoff() time.Duration { + if dlv.next.IsZero() { + return 0 } - - // Pop from backlog. - dlv := w.backlog[0] - - // Shift backlog down by one. - copy(w.backlog, w.backlog[1:]) - w.backlog = w.backlog[:len(w.backlog)-1] - - return dlv -} - -// pushBacklog pushes the given delivery to backlog. -func (w *Worker) pushBacklog(dlv *Delivery) { - w.backlog = append(w.backlog, dlv) -} - -// sortDeliveries sorts deliveries according -// to when is the first requiring re-attempt. -func sortDeliveries(d []*Delivery) { - slices.SortFunc(d, func(a, b *Delivery) int { - const k = +1 - switch { - case a.next.Before(b.next): - return +k - case b.next.Before(a.next): - return -k - default: - return 0 - } - }) + return time.Until(dlv.next) } diff --git a/internal/transport/delivery/delivery_test.go b/internal/transport/delivery/delivery_test.go index 48831f098..e9eaf8fd1 100644 --- a/internal/transport/delivery/delivery_test.go +++ b/internal/transport/delivery/delivery_test.go @@ -1,203 +1,134 @@ +// GoToSocial +// Copyright (C) GoToSocial Authors admin@gotosocial.org +// SPDX-License-Identifier: AGPL-3.0-or-later +// +// This program is free software: you can redistribute it and/or modify +// it under the terms of the GNU Affero General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// This program is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Affero General Public License for more details. +// +// You should have received a copy of the GNU Affero General Public License +// along with this program. If not, see <http://www.gnu.org/licenses/>. + package delivery_test import ( - "fmt" + "bytes" + "encoding/json" "io" - "math/rand" - "net" "net/http" - "strconv" - "strings" "testing" - "codeberg.org/gruf/go-byteutil" - "github.com/superseriousbusiness/gotosocial/internal/config" + "github.com/stretchr/testify/assert" "github.com/superseriousbusiness/gotosocial/internal/httpclient" - "github.com/superseriousbusiness/gotosocial/internal/queue" "github.com/superseriousbusiness/gotosocial/internal/transport/delivery" ) -func TestDeliveryWorkerPool(t *testing.T) { - for _, i := range []int{1, 2, 4, 8, 16, 32} { - t.Run("size="+strconv.Itoa(i), func(t *testing.T) { - testDeliveryWorkerPool(t, i, generateInput(100*i)) - }) - } -} - -func testDeliveryWorkerPool(t *testing.T, sz int, input []*testrequest) { - wp := new(delivery.WorkerPool) - wp.Init(httpclient.New(httpclient.Config{ - AllowRanges: config.MustParseIPPrefixes([]string{ - "127.0.0.0/8", +var deliveryCases = []struct { + msg delivery.Delivery + data []byte +}{ + { + msg: delivery.Delivery{ + PubKeyID: "https://google.com/users/bigboy#pubkey", + ActorID: "https://google.com/users/bigboy", + ObjectID: "https://google.com/users/bigboy/follow/1", + TargetID: "https://askjeeves.com/users/smallboy", + Request: toRequest("POST", "https://askjeeves.com/users/smallboy/inbox", []byte("data!")), + }, + data: toJSON(map[string]any{ + "pub_key_id": "https://google.com/users/bigboy#pubkey", + "actor_id": "https://google.com/users/bigboy", + "object_id": "https://google.com/users/bigboy/follow/1", + "target_id": "https://askjeeves.com/users/smallboy", + "method": "POST", + "url": "https://askjeeves.com/users/smallboy/inbox", + "body": []byte("data!"), + // "header": map[string][]string{}, + }), + }, + { + msg: delivery.Delivery{ + Request: toRequest("GET", "https://google.com", []byte("uwu im just a wittle seawch engwin")), + }, + data: toJSON(map[string]any{ + "method": "GET", + "url": "https://google.com", + "body": []byte("uwu im just a wittle seawch engwin"), + // "header": map[string][]string{}, }), - })) - wp.Start(sz) - defer wp.Stop() - test(t, &wp.Queue, input) + }, } -func test( - t *testing.T, - queue *queue.StructQueue[*delivery.Delivery], - input []*testrequest, -) { - expect := make(chan *testrequest) - errors := make(chan error) - - // Prepare an HTTP test handler that ensures expected delivery is received. - handler := http.HandlerFunc(func(rw http.ResponseWriter, r *http.Request) { - errors <- (<-expect).Equal(r) - }) - - // Start new HTTP test server listener. - l, err := net.Listen("tcp", "127.0.0.1:0") - if err != nil { - t.Fatal(err) - } - defer l.Close() - - // Start the HTTP server. - // - // specifically not using httptest.Server{} here as httptest - // links that server with its own http.Client{}, whereas we're - // using an httpclient.Client{} (well, delivery routine is). - srv := new(http.Server) - srv.Addr = "http://" + l.Addr().String() - srv.Handler = handler - go srv.Serve(l) - defer srv.Close() - - // Range over test input. - for _, test := range input { - - // Generate req for input. - req := test.Generate(srv.Addr) - r := httpclient.WrapRequest(req) - - // Wrap the request in delivery. - dlv := new(delivery.Delivery) - dlv.Request = r - - // Enqueue delivery! - queue.Push(dlv) - expect <- test - - // Wait for errors from handler. - if err := <-errors; err != nil { - t.Error(err) +func TestSerializeDelivery(t *testing.T) { + for _, test := range deliveryCases { + // Serialize test message to blob. + data, err := test.msg.Serialize() + if err != nil { + t.Fatal(err) } - } -} - -type testrequest struct { - method string - uri string - body []byte -} -// generateInput generates 'n' many testrequest cases. -func generateInput(n int) []*testrequest { - tests := make([]*testrequest, n) - for i := range tests { - tests[i] = new(testrequest) - tests[i].method = randomMethod() - tests[i].uri = randomURI() - tests[i].body = randomBody(tests[i].method) + // Check that serialized JSON data is as expected. + assert.JSONEq(t, string(test.data), string(data)) } - return tests } -var methods = []string{ - http.MethodConnect, - http.MethodDelete, - http.MethodGet, - http.MethodHead, - http.MethodOptions, - http.MethodPatch, - http.MethodPost, - http.MethodPut, - http.MethodTrace, -} +func TestDeserializeDelivery(t *testing.T) { + for _, test := range deliveryCases { + var msg delivery.Delivery -// randomMethod generates a random http method. -func randomMethod() string { - return methods[rand.Intn(len(methods))] -} + // Deserialize test message blob. + err := msg.Deserialize(test.data) + if err != nil { + t.Fatal(err) + } -// randomURI generates a random http uri. -func randomURI() string { - n := rand.Intn(5) - p := make([]string, n) - for i := range p { - p[i] = strconv.Itoa(rand.Int()) + // Check that delivery fields are as expected. + assert.Equal(t, test.msg.PubKeyID, msg.PubKeyID) + assert.Equal(t, test.msg.ActorID, msg.ActorID) + assert.Equal(t, test.msg.ObjectID, msg.ObjectID) + assert.Equal(t, test.msg.TargetID, msg.TargetID) + assert.Equal(t, test.msg.Request.Method, msg.Request.Method) + assert.Equal(t, test.msg.Request.URL, msg.Request.URL) + assert.Equal(t, readBody(test.msg.Request.Body), readBody(msg.Request.Body)) } - return "/" + strings.Join(p, "/") } -// randomBody generates a random http body DEPENDING on method. -func randomBody(method string) []byte { - if requiresBody(method) { - return []byte(method + " " + randomURI()) +// toRequest creates httpclient.Request from HTTP method, URL and body data. +func toRequest(method string, url string, body []byte) httpclient.Request { + var rbody io.Reader + if body != nil { + rbody = bytes.NewReader(body) } - return nil -} - -// requiresBody returns whether method requires body. -func requiresBody(method string) bool { - switch method { - case http.MethodPatch, - http.MethodPost, - http.MethodPut: - return true - default: - return false + req, err := http.NewRequest(method, url, rbody) + if err != nil { + panic(err) } + return httpclient.WrapRequest(req) } -// Generate will generate a real http.Request{} from test data. -func (t *testrequest) Generate(addr string) *http.Request { - var body io.ReadCloser - if t.body != nil { - var b byteutil.ReadNopCloser - b.Reset(t.body) - body = &b +// readBody reads the content of body io.ReadCloser into memory as byte slice. +func readBody(r io.ReadCloser) []byte { + if r == nil { + return nil } - req, err := http.NewRequest(t.method, addr+t.uri, body) + b, err := io.ReadAll(r) if err != nil { panic(err) } - return req + return b } -// Equal checks if request matches receiving test request. -func (t *testrequest) Equal(r *http.Request) error { - // Ensure methods match. - if t.method != r.Method { - return fmt.Errorf("differing request methods: t=%q r=%q", t.method, r.Method) - } - - // Ensure request URIs match. - if t.uri != r.URL.RequestURI() { - return fmt.Errorf("differing request urls: t=%q r=%q", t.uri, r.URL.RequestURI()) - } - - // Ensure body cases match. - if requiresBody(t.method) { - - // Read request into memory. - b, err := io.ReadAll(r.Body) - if err != nil { - return fmt.Errorf("error reading request body: %v", err) - } - - // Compare the request bodies. - st := strings.TrimSpace(string(t.body)) - sr := strings.TrimSpace(string(b)) - if st != sr { - return fmt.Errorf("differing request bodies: t=%q r=%q", st, sr) - } +// toJSON marshals input type as JSON data. +func toJSON(a any) []byte { + b, err := json.Marshal(a) + if err != nil { + panic(err) } - - return nil + return b } diff --git a/internal/transport/delivery/worker.go b/internal/transport/delivery/worker.go new file mode 100644 index 000000000..1ed974e84 --- /dev/null +++ b/internal/transport/delivery/worker.go @@ -0,0 +1,298 @@ +// GoToSocial +// Copyright (C) GoToSocial Authors admin@gotosocial.org +// SPDX-License-Identifier: AGPL-3.0-or-later +// +// This program is free software: you can redistribute it and/or modify +// it under the terms of the GNU Affero General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// This program is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Affero General Public License for more details. +// +// You should have received a copy of the GNU Affero General Public License +// along with this program. If not, see <http://www.gnu.org/licenses/>. + +package delivery + +import ( + "context" + "slices" + "time" + + "codeberg.org/gruf/go-runners" + "codeberg.org/gruf/go-structr" + "github.com/superseriousbusiness/gotosocial/internal/gtscontext" + "github.com/superseriousbusiness/gotosocial/internal/httpclient" + "github.com/superseriousbusiness/gotosocial/internal/log" + "github.com/superseriousbusiness/gotosocial/internal/queue" + "github.com/superseriousbusiness/gotosocial/internal/util" +) + +// WorkerPool wraps multiple Worker{}s in +// a singular struct for easy multi start/stop. +type WorkerPool struct { + + // Client defines httpclient.Client{} + // passed to each of delivery pool Worker{}s. + Client *httpclient.Client + + // Queue is the embedded queue.StructQueue{} + // passed to each of delivery pool Worker{}s. + Queue queue.StructQueue[*Delivery] + + // internal fields. + workers []*Worker +} + +// Init will initialize the Worker{} pool +// with given http client, request queue to pull +// from and number of delivery workers to spawn. +func (p *WorkerPool) Init(client *httpclient.Client) { + p.Client = client + p.Queue.Init(structr.QueueConfig[*Delivery]{ + Indices: []structr.IndexConfig{ + {Fields: "ActorID", Multiple: true}, + {Fields: "ObjectID", Multiple: true}, + {Fields: "TargetID", Multiple: true}, + }, + }) +} + +// Start will attempt to start 'n' Worker{}s. +func (p *WorkerPool) Start(n int) { + // Check whether workers are + // set (is already running). + ok := (len(p.workers) > 0) + if ok { + return + } + + // Allocate new workers slice. + p.workers = make([]*Worker, n) + for i := range p.workers { + + // Allocate new Worker{}. + p.workers[i] = new(Worker) + p.workers[i].Client = p.Client + p.workers[i].Queue = &p.Queue + + // Attempt to start worker. + // Return bool not useful + // here, as true = started, + // false = already running. + _ = p.workers[i].Start() + } +} + +// Stop will attempt to stop contained Worker{}s. +func (p *WorkerPool) Stop() { + // Check whether workers are + // set (is currently running). + ok := (len(p.workers) == 0) + if ok { + return + } + + // Stop all running workers. + for i := range p.workers { + + // return bool not useful + // here, as true = stopped, + // false = never running. + _ = p.workers[i].Stop() + } + + // Unset workers slice. + p.workers = p.workers[:0] +} + +// Worker wraps an httpclient.Client{} to feed +// from queue.StructQueue{} for ActivityPub reqs +// to deliver. It does so while prioritizing new +// queued requests over backlogged retries. +type Worker struct { + + // Client is the httpclient.Client{} that + // delivery worker will use for requests. + Client *httpclient.Client + + // Queue is the Delivery{} message queue + // that delivery worker will feed from. + Queue *queue.StructQueue[*Delivery] + + // internal fields. + backlog []*Delivery + service runners.Service +} + +// Start will attempt to start the Worker{}. +func (w *Worker) Start() bool { + return w.service.GoRun(w.run) +} + +// Stop will attempt to stop the Worker{}. +func (w *Worker) Stop() bool { + return w.service.Stop() +} + +// run wraps process to restart on any panic. +func (w *Worker) run(ctx context.Context) { + if w.Client == nil || w.Queue == nil { + panic("not yet initialized") + } + log.Infof(ctx, "%p: starting worker", w) + defer log.Infof(ctx, "%p: stopped worker", w) + util.Must(func() { w.process(ctx) }) +} + +// process is the main delivery worker processing routine. +func (w *Worker) process(ctx context.Context) bool { + if w.Client == nil || w.Queue == nil { + // we perform this check here just + // to ensure the compiler knows these + // variables aren't nil in the loop, + // even if already checked by caller. + panic("not yet initialized") + } + +loop: + for { + // Get next delivery. + dlv, ok := w.next(ctx) + if !ok { + return true + } + + // Check whether backoff required. + const min = 100 * time.Millisecond + if d := dlv.backoff(); d > min { + + // Start backoff sleep timer. + backoff := time.NewTimer(d) + + select { + case <-ctx.Done(): + // Main ctx + // cancelled. + backoff.Stop() + return true + + case <-w.Queue.Wait(): + // A new message was + // queued, re-add this + // to backlog + retry. + w.pushBacklog(dlv) + backoff.Stop() + continue loop + + case <-backoff.C: + // success! + } + } + + // Attempt delivery of AP request. + rsp, retry, err := w.Client.DoOnce( + &dlv.Request, + ) + + if err == nil { + // Ensure body closed. + _ = rsp.Body.Close() + continue loop + } + + if !retry { + // Drop deliveries when no + // retry requested, or they + // reached max (either). + continue loop + } + + // Determine next delivery attempt. + backoff := dlv.Request.BackOff() + dlv.next = time.Now().Add(backoff) + + // Push to backlog. + w.pushBacklog(dlv) + } +} + +// next gets the next available delivery, blocking until available if necessary. +func (w *Worker) next(ctx context.Context) (*Delivery, bool) { +loop: + for { + // Try pop next queued. + dlv, ok := w.Queue.Pop() + + if !ok { + // Check the backlog. + if len(w.backlog) > 0 { + + // Sort by 'next' time. + sortDeliveries(w.backlog) + + // Pop next delivery. + dlv := w.popBacklog() + + return dlv, true + } + + select { + // Backlog is empty, we MUST + // block until next enqueued. + case <-w.Queue.Wait(): + continue loop + + // Worker was stopped. + case <-ctx.Done(): + return nil, false + } + } + + // Replace request context for worker state canceling. + ctx := gtscontext.WithValues(ctx, dlv.Request.Context()) + dlv.Request.Request = dlv.Request.Request.WithContext(ctx) + + return dlv, true + } +} + +// popBacklog pops next available from the backlog. +func (w *Worker) popBacklog() *Delivery { + if len(w.backlog) == 0 { + return nil + } + + // Pop from backlog. + dlv := w.backlog[0] + + // Shift backlog down by one. + copy(w.backlog, w.backlog[1:]) + w.backlog = w.backlog[:len(w.backlog)-1] + + return dlv +} + +// pushBacklog pushes the given delivery to backlog. +func (w *Worker) pushBacklog(dlv *Delivery) { + w.backlog = append(w.backlog, dlv) +} + +// sortDeliveries sorts deliveries according +// to when is the first requiring re-attempt. +func sortDeliveries(d []*Delivery) { + slices.SortFunc(d, func(a, b *Delivery) int { + const k = +1 + switch { + case a.next.Before(b.next): + return +k + case b.next.Before(a.next): + return -k + default: + return 0 + } + }) +} diff --git a/internal/transport/delivery/worker_test.go b/internal/transport/delivery/worker_test.go new file mode 100644 index 000000000..936ce6e1d --- /dev/null +++ b/internal/transport/delivery/worker_test.go @@ -0,0 +1,220 @@ +// GoToSocial +// Copyright (C) GoToSocial Authors admin@gotosocial.org +// SPDX-License-Identifier: AGPL-3.0-or-later +// +// This program is free software: you can redistribute it and/or modify +// it under the terms of the GNU Affero General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// This program is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Affero General Public License for more details. +// +// You should have received a copy of the GNU Affero General Public License +// along with this program. If not, see <http://www.gnu.org/licenses/>. + +package delivery_test + +import ( + "fmt" + "io" + "math/rand" + "net" + "net/http" + "strconv" + "strings" + "testing" + + "codeberg.org/gruf/go-byteutil" + "github.com/superseriousbusiness/gotosocial/internal/config" + "github.com/superseriousbusiness/gotosocial/internal/httpclient" + "github.com/superseriousbusiness/gotosocial/internal/queue" + "github.com/superseriousbusiness/gotosocial/internal/transport/delivery" +) + +func TestDeliveryWorkerPool(t *testing.T) { + for _, i := range []int{1, 2, 4, 8, 16, 32} { + t.Run("size="+strconv.Itoa(i), func(t *testing.T) { + testDeliveryWorkerPool(t, i, generateInput(100*i)) + }) + } +} + +func testDeliveryWorkerPool(t *testing.T, sz int, input []*testrequest) { + wp := new(delivery.WorkerPool) + wp.Init(httpclient.New(httpclient.Config{ + AllowRanges: config.MustParseIPPrefixes([]string{ + "127.0.0.0/8", + }), + })) + wp.Start(sz) + defer wp.Stop() + test(t, &wp.Queue, input) +} + +func test( + t *testing.T, + queue *queue.StructQueue[*delivery.Delivery], + input []*testrequest, +) { + expect := make(chan *testrequest) + errors := make(chan error) + + // Prepare an HTTP test handler that ensures expected delivery is received. + handler := http.HandlerFunc(func(rw http.ResponseWriter, r *http.Request) { + errors <- (<-expect).Equal(r) + }) + + // Start new HTTP test server listener. + l, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + defer l.Close() + + // Start the HTTP server. + // + // specifically not using httptest.Server{} here as httptest + // links that server with its own http.Client{}, whereas we're + // using an httpclient.Client{} (well, delivery routine is). + srv := new(http.Server) + srv.Addr = "http://" + l.Addr().String() + srv.Handler = handler + go srv.Serve(l) + defer srv.Close() + + // Range over test input. + for _, test := range input { + + // Generate req for input. + req := test.Generate(srv.Addr) + r := httpclient.WrapRequest(req) + + // Wrap the request in delivery. + dlv := new(delivery.Delivery) + dlv.Request = r + + // Enqueue delivery! + queue.Push(dlv) + expect <- test + + // Wait for errors from handler. + if err := <-errors; err != nil { + t.Error(err) + } + } +} + +type testrequest struct { + method string + uri string + body []byte +} + +// generateInput generates 'n' many testrequest cases. +func generateInput(n int) []*testrequest { + tests := make([]*testrequest, n) + for i := range tests { + tests[i] = new(testrequest) + tests[i].method = randomMethod() + tests[i].uri = randomURI() + tests[i].body = randomBody(tests[i].method) + } + return tests +} + +var methods = []string{ + http.MethodConnect, + http.MethodDelete, + http.MethodGet, + http.MethodHead, + http.MethodOptions, + http.MethodPatch, + http.MethodPost, + http.MethodPut, + http.MethodTrace, +} + +// randomMethod generates a random http method. +func randomMethod() string { + return methods[rand.Intn(len(methods))] +} + +// randomURI generates a random http uri. +func randomURI() string { + n := rand.Intn(5) + p := make([]string, n) + for i := range p { + p[i] = strconv.Itoa(rand.Int()) + } + return "/" + strings.Join(p, "/") +} + +// randomBody generates a random http body DEPENDING on method. +func randomBody(method string) []byte { + if requiresBody(method) { + return []byte(method + " " + randomURI()) + } + return nil +} + +// requiresBody returns whether method requires body. +func requiresBody(method string) bool { + switch method { + case http.MethodPatch, + http.MethodPost, + http.MethodPut: + return true + default: + return false + } +} + +// Generate will generate a real http.Request{} from test data. +func (t *testrequest) Generate(addr string) *http.Request { + var body io.ReadCloser + if t.body != nil { + var b byteutil.ReadNopCloser + b.Reset(t.body) + body = &b + } + req, err := http.NewRequest(t.method, addr+t.uri, body) + if err != nil { + panic(err) + } + return req +} + +// Equal checks if request matches receiving test request. +func (t *testrequest) Equal(r *http.Request) error { + // Ensure methods match. + if t.method != r.Method { + return fmt.Errorf("differing request methods: t=%q r=%q", t.method, r.Method) + } + + // Ensure request URIs match. + if t.uri != r.URL.RequestURI() { + return fmt.Errorf("differing request urls: t=%q r=%q", t.uri, r.URL.RequestURI()) + } + + // Ensure body cases match. + if requiresBody(t.method) { + + // Read request into memory. + b, err := io.ReadAll(r.Body) + if err != nil { + return fmt.Errorf("error reading request body: %v", err) + } + + // Compare the request bodies. + st := strings.TrimSpace(string(t.body)) + sr := strings.TrimSpace(string(b)) + if st != sr { + return fmt.Errorf("differing request bodies: t=%q r=%q", st, sr) + } + } + + return nil +} |