diff options
Diffstat (limited to 'internal/transport')
| -rw-r--r-- | internal/transport/controller.go | 16 | ||||
| -rw-r--r-- | internal/transport/deliver.go | 212 | ||||
| -rw-r--r-- | internal/transport/delivery/delivery.go | 323 | ||||
| -rw-r--r-- | internal/transport/delivery/delivery_test.go | 205 | ||||
| -rw-r--r-- | internal/transport/transport.go | 7 |
5 files changed, 676 insertions, 87 deletions
diff --git a/internal/transport/controller.go b/internal/transport/controller.go index 891a24495..519298d8e 100644 --- a/internal/transport/controller.go +++ b/internal/transport/controller.go @@ -28,7 +28,6 @@ import ( "io" "net/http" "net/url" - "runtime" "codeberg.org/gruf/go-byteutil" "codeberg.org/gruf/go-cache/v3" @@ -56,24 +55,16 @@ type controller struct { client pub.HttpClient trspCache cache.TTLCache[string, *transport] userAgent string - senders int // no. concurrent batch delivery routines. } // NewController returns an implementation of the Controller interface for creating new transports func NewController(state *state.State, federatingDB federatingdb.DB, clock pub.Clock, client pub.HttpClient) Controller { var ( - host = config.GetHost() - proto = config.GetProtocol() - version = config.GetSoftwareVersion() - senderMultiplier = config.GetAdvancedSenderMultiplier() + host = config.GetHost() + proto = config.GetProtocol() + version = config.GetSoftwareVersion() ) - senders := senderMultiplier * runtime.GOMAXPROCS(0) - if senders < 1 { - // Clamp senders to 1. - senders = 1 - } - c := &controller{ state: state, fedDB: federatingDB, @@ -81,7 +72,6 @@ func NewController(state *state.State, federatingDB federatingdb.DB, clock pub.C client: client, trspCache: cache.NewTTL[string, *transport](0, 100, 0), userAgent: fmt.Sprintf("gotosocial/%s (+%s://%s)", version, proto, host), - senders: senders, } return c diff --git a/internal/transport/deliver.go b/internal/transport/deliver.go index fe4d04582..a7e73465d 100644 --- a/internal/transport/deliver.go +++ b/internal/transport/deliver.go @@ -19,118 +19,188 @@ package transport import ( "context" + "encoding/json" "net/http" "net/url" - "sync" "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" "github.com/superseriousbusiness/gotosocial/internal/gtserror" + "github.com/superseriousbusiness/gotosocial/internal/httpclient" + "github.com/superseriousbusiness/gotosocial/internal/transport/delivery" ) -func (t *transport) BatchDeliver(ctx context.Context, b []byte, recipients []*url.URL) error { +func (t *transport) BatchDeliver(ctx context.Context, obj map[string]interface{}, recipients []*url.URL) error { var ( - // errs accumulates errors received during - // attempted delivery by deliverer routines. - errs gtserror.MultiError - - // wait blocks until all sender - // routines have returned. - wait sync.WaitGroup + // accumulated delivery reqs. + reqs []*delivery.Delivery - // mutex protects 'recipients' and - // 'errs' for concurrent access. - mutex sync.Mutex + // accumulated preparation errs. + errs gtserror.MultiError // Get current instance host info. domain = config.GetAccountDomain() host = config.GetHost() ) - // Block on expect no. senders. - wait.Add(t.controller.senders) - - for i := 0; i < t.controller.senders; i++ { - go func() { - // Mark returned. - defer wait.Done() - - for { - // Acquire lock. - mutex.Lock() - - if len(recipients) == 0 { - // Reached end. - mutex.Unlock() - return - } - - // Pop next recipient. - i := len(recipients) - 1 - to := recipients[i] - recipients = recipients[:i] - - // Done with lock. - mutex.Unlock() - - // Skip delivery to recipient if it is "us". - if to.Host == host || to.Host == domain { - continue - } - - // Attempt to deliver data to recipient. - if err := t.deliver(ctx, b, to); err != nil { - mutex.Lock() // safely append err to accumulator. - errs.Appendf("error delivering to %s: %w", to, err) - mutex.Unlock() - } - } - }() + // Marshal object as JSON. + b, err := json.Marshal(obj) + if err != nil { + return gtserror.Newf("error marshaling json: %w", err) + } + + // Extract object IDs. + actID := getActorID(obj) + objID := getObjectID(obj) + tgtID := getTargetID(obj) + + for _, to := range recipients { + // Skip delivery to recipient if it is "us". + if to.Host == host || to.Host == domain { + continue + } + + // Prepare http client request. + req, err := t.prepare(ctx, + actID, + objID, + tgtID, + b, + to, + ) + if err != nil { + errs.Append(err) + continue + } + + // Append to request queue. + reqs = append(reqs, req) } - // Wait for finish. - wait.Wait() + // Push prepared request list to the delivery queue. + t.controller.state.Workers.Delivery.Queue.Push(reqs...) // Return combined err. return errs.Combine() } -func (t *transport) Deliver(ctx context.Context, b []byte, to *url.URL) error { +func (t *transport) Deliver(ctx context.Context, obj map[string]interface{}, to *url.URL) error { // if 'to' host is our own, skip as we don't need to deliver to ourselves... if to.Host == config.GetHost() || to.Host == config.GetAccountDomain() { return nil } - // Deliver data to recipient. - return t.deliver(ctx, b, to) + // Marshal object as JSON. + b, err := json.Marshal(obj) + if err != nil { + return gtserror.Newf("error marshaling json: %w", err) + } + + // Prepare http client request. + req, err := t.prepare(ctx, + getActorID(obj), + getObjectID(obj), + getTargetID(obj), + b, + to, + ) + if err != nil { + return err + } + + // Push prepared request to the delivery queue. + t.controller.state.Workers.Delivery.Queue.Push(req) + + return nil } -func (t *transport) deliver(ctx context.Context, b []byte, to *url.URL) error { +// prepare will prepare a POST http.Request{} +// to recipient at 'to', wrapping in a queued +// request object with signing function. +func (t *transport) prepare( + ctx context.Context, + actorID string, + objectID string, + targetID string, + data []byte, + to *url.URL, +) ( + *delivery.Delivery, + error, +) { url := to.String() - // Use rewindable bytes reader for body. + // Use rewindable reader for body. var body byteutil.ReadNopCloser - body.Reset(b) + body.Reset(data) + + // Prepare POST signer. + sign := t.signPOST(data) - req, err := http.NewRequestWithContext(ctx, "POST", url, &body) + // 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) if err != nil { - return err + return nil, gtserror.Newf("error preparing request: %w", err) } - req.Header.Add("Content-Type", string(apiutil.AppActivityLDJSON)) - req.Header.Add("Accept-Charset", "utf-8") + // Set the standard ActivityPub content-type + charset headers. + r.Header.Add("Content-Type", string(apiutil.AppActivityLDJSON)) + r.Header.Add("Accept-Charset", "utf-8") - rsp, err := t.POST(req, b) - if err != nil { - return err + // Validate the request before queueing for delivery. + if err := httpclient.ValidateRequest(r); err != nil { + return nil, err + } + + return &delivery.Delivery{ + ActorID: actorID, + ObjectID: objectID, + TargetID: targetID, + Request: httpclient.WrapRequest(r), + }, nil +} + +// getObjectID extracts an object ID from 'serialized' ActivityPub object map. +func getObjectID(obj map[string]interface{}) string { + switch t := obj["object"].(type) { + case string: + return t + case map[string]interface{}: + id, _ := t["id"].(string) + return id + default: + return "" } - defer rsp.Body.Close() +} - if code := rsp.StatusCode; code != http.StatusOK && - code != http.StatusCreated && code != http.StatusAccepted { - return gtserror.NewFromResponse(rsp) +// getActorID extracts an actor ID from 'serialized' ActivityPub object map. +func getActorID(obj map[string]interface{}) string { + switch t := obj["actor"].(type) { + case string: + return t + case map[string]interface{}: + id, _ := t["id"].(string) + return id + default: + return "" } +} - return nil +// getTargetID extracts a target ID from 'serialized' ActivityPub object map. +func getTargetID(obj map[string]interface{}) string { + switch t := obj["target"].(type) { + case string: + return t + case map[string]interface{}: + id, _ := t["id"].(string) + return id + default: + return "" + } } diff --git a/internal/transport/delivery/delivery.go b/internal/transport/delivery/delivery.go new file mode 100644 index 000000000..27281399f --- /dev/null +++ b/internal/transport/delivery/delivery.go @@ -0,0 +1,323 @@ +// 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" +) + +// Delivery wraps an httpclient.Request{} +// to add ActivityPub ID IRI fields of the +// outgoing activity, so that deliveries may +// be indexed (and so, dropped from queue) +// by any of these possible ID IRIs. +type Delivery struct { + + // ActorID contains the ActivityPub + // actor ID IRI (if any) of the activity + // being sent out by this request. + ActorID string + + // ObjectID contains the ActivityPub + // object ID IRI (if any) of the activity + // being sent out by this request. + ObjectID string + + // TargetID contains the ActivityPub + // target ID IRI (if any) of the activity + // being sent out by this request. + TargetID string + + // Request is the prepared (+ wrapped) + // httpclient.Client{} request that + // constitutes this ActivtyPub delivery. + Request httpclient.Request + + // internal fields. + next time.Time +} + +func (dlv *Delivery) backoff() time.Duration { + if dlv.next.IsZero() { + return 0 + } + return time.Until(dlv.next) +} + +// 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) (ok bool) { + if ok = (len(p.workers) == 0); ok { + p.workers = make([]*Worker, n) + for i := range p.workers { + p.workers[i] = new(Worker) + p.workers[i].Client = p.Client + p.workers[i].Queue = &p.Queue + ok = p.workers[i].Start() && ok + } + } + return +} + +// Stop will attempt to stop contained Worker{}s. +func (p *WorkerPool) Stop() (ok bool) { + if ok = (len(p.workers) > 0); ok { + for i := range p.workers { + ok = p.workers[i].Stop() && ok + p.workers[i] = nil + } + p.workers = p.workers[:0] + } + return +} + +// 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: started delivery worker", w) + defer log.Infof(ctx, "%p: stopped delivery worker", w) + for returned := false; !returned; { + func() { + defer func() { + if r := recover(); r != nil { + log.Errorf(ctx, "recovered panic: %v", r) + } + }() + w.process(ctx) + returned = true + }() + } +} + +// process is the main delivery worker processing routine. +func (w *Worker) process(ctx context.Context) { + 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 + } + + // 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 + + 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/delivery_test.go b/internal/transport/delivery/delivery_test.go new file mode 100644 index 000000000..852c6f6f3 --- /dev/null +++ b/internal/transport/delivery/delivery_test.go @@ -0,0 +1,205 @@ +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", + }), + })) + if !wp.Start(sz) { + t.Fatal("failed starting pool") + } + 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 +} diff --git a/internal/transport/transport.go b/internal/transport/transport.go index 3ae5c8967..110c19b3d 100644 --- a/internal/transport/transport.go +++ b/internal/transport/transport.go @@ -51,10 +51,10 @@ type Transport interface { POST(*http.Request, []byte) (*http.Response, error) // Deliver sends an ActivityStreams object. - Deliver(ctx context.Context, b []byte, to *url.URL) error + Deliver(ctx context.Context, obj map[string]interface{}, to *url.URL) error // BatchDeliver sends an ActivityStreams object to multiple recipients. - BatchDeliver(ctx context.Context, b []byte, recipients []*url.URL) error + BatchDeliver(ctx context.Context, obj map[string]interface{}, recipients []*url.URL) error /* GET functions @@ -77,7 +77,8 @@ type Transport interface { Finger(ctx context.Context, targetUsername string, targetDomain string) ([]byte, error) } -// transport implements the Transport interface. +// transport implements +// the Transport interface. type transport struct { controller *controller pubKeyID string |