From 3b14d0b455e37c097dbe0e49671d0766eb47a665 Mon Sep 17 00:00:00 2001 From: Mark Mennell Date: Sat, 10 Oct 2026 15:00:50 +0800 Subject: [PATCH] Count attachments and use the accept time when recording receive usage fmsgd recorded a received message's usage with only its body's wire size and the sender's timestamp, so attachments were never counted and a backdated message fell outside the per-day window. Usage is now the stored size, the body plus every attachment, each decompressed if it was sent compressed. It's timestamped when this host accepted the message. The receive limit check uses the same stored size, so a message whose attachments take a recipient over a limit is rejected with 101 (user full), as one with a large body already was. Co-Authored-By: Claude Opus 5.5 --- README.md | 4 ++ cmd/fmsgd/host.go | 66 +++++++++++++++----- cmd/fmsgd/id.go | 6 +- cmd/fmsgd/usage_test.go | 130 ++++++++++++++++++++++++++++++++++++++++ 4 files changed, 188 insertions(+), 18 deletions(-) create mode 100644 cmd/fmsgd/usage_test.go diff --git a/README.md b/README.md index c58dfb0..3ea9db9 100644 --- a/README.md +++ b/README.md @@ -42,6 +42,10 @@ Requires Go 1.27 or newer. +## Quotas and usage + +For each local recipient of a message from another host, fmsgd asks fmsgid for the address's receive limits and usage, and rejects the recipient with `101` (user full) if one more message of this size would exceed its messages per day, size per day or total size (`-1` is unlimited). The size is the stored size: the message body plus every attachment, decompressed. Once the message is stored, fmsgd records that size for each recipient that accepted it with `POST /fmsgid/recv`, timestamped when this host accepted the message (not the sender's timestamp), so a backdated message still counts towards today. fmsgd doesn't record send usage or mail between local addresses: the client that sends and delivers locally (such as fmsg-webapi) does. + ## Running An up and running [fmsg Id API](https://github.com/markmnl/fmsgid) needs to be reachable by fmsgd to know users and their quotas for this fmsgd service. See also [fmsg-docker](https://github.com/markmnl/fmsg-docker) - a docker compose stack for a fmsg host including fmsgid, fmsg-webpi and fmsgd. diff --git a/cmd/fmsgd/host.go b/cmd/fmsgd/host.go index 1b8dcb9..53e8635 100644 --- a/cmd/fmsgd/host.go +++ b/cmd/fmsgd/host.go @@ -1253,21 +1253,48 @@ func validateMsgRecvForAddr(h *FMsgHeader, addr *FMsgAddress, msgHash []byte) (c return RejectCodeUserNotAccepting, nil } - // check user limits - if detail.LimitRecvCountPer1d > -1 && detail.RecvCountPer1d+1 > detail.LimitRecvCountPer1d { - log.Printf("WARN: Message rejected: RecvCountPer1d would exceed LimitRecvCountPer1d %d", detail.LimitRecvCountPer1d) + // check user limits against the size that will be recorded once accepted + if exceeded := recvLimitExceeded(detail, storedSize(h)); exceeded != "" { + log.Printf("WARN: Message rejected: %s", exceeded) return RejectCodeUserFull, nil } - if detail.LimitRecvSizePer1d > -1 && detail.RecvSizePer1d+int64(h.Size) > detail.LimitRecvSizePer1d { - log.Printf("WARN: Message rejected: RecvSizePer1d would exceed LimitRecvSizePer1d %d", detail.LimitRecvSizePer1d) - return RejectCodeUserFull, nil + + return RejectCodeAccept, nil +} + +// storedSize is the size a received message takes in storage, and so the size +// recorded as the recipient's usage: the message body plus every attachment, +// each expanded (stored decompressed) when sent compressed. Limits are checked +// against this same size before the message is accepted. +func storedSize(h *FMsgHeader) int64 { + size := int64(h.Size) + if h.Flags&FlagDeflate != 0 { + size = int64(h.ExpandedSize) } - if detail.LimitRecvSizeTotal > -1 && detail.RecvSizeTotal+int64(h.Size) > detail.LimitRecvSizeTotal { - log.Printf("WARN: Message rejected: RecvSizeTotal would exceed LimitRecvSizeTotal %d", detail.LimitRecvSizeTotal) - return RejectCodeUserFull, nil + for _, a := range h.Attachments { + if a.Flags&(1<<1) != 0 { + size += int64(a.ExpandedSize) + } else { + size += int64(a.Size) + } } + return size +} - return RejectCodeAccept, nil +// recvLimitExceeded checks one more message of size bytes against an +// address's receive limits (a limit of -1 is unlimited). It returns which +// limit would be exceeded, or "" when the message fits. +func recvLimitExceeded(detail *AddressDetail, size int64) string { + if detail.LimitRecvCountPer1d > -1 && detail.RecvCountPer1d+1 > detail.LimitRecvCountPer1d { + return fmt.Sprintf("RecvCountPer1d would exceed LimitRecvCountPer1d %d", detail.LimitRecvCountPer1d) + } + if detail.LimitRecvSizePer1d > -1 && detail.RecvSizePer1d+size > detail.LimitRecvSizePer1d { + return fmt.Sprintf("RecvSizePer1d would exceed LimitRecvSizePer1d %d", detail.LimitRecvSizePer1d) + } + if detail.LimitRecvSizeTotal > -1 && detail.RecvSizeTotal+size > detail.LimitRecvSizeTotal { + return fmt.Sprintf("RecvSizeTotal would exceed LimitRecvSizeTotal %d", detail.LimitRecvSizeTotal) + } + return "" } // uniqueFilepath generates a unique file path in the given directory, @@ -1515,13 +1542,22 @@ func storeAcceptedMessage(h *FMsgHeader, codes []byte, acceptedTo []FMsgAddress, return false } - allAccepted := append(acceptedTo, acceptedAddTo...) - for i := range allAccepted { - if err := postMsgStatRecv(&allAccepted[i], h.Timestamp, int(h.Size)); err != nil { - log.Printf("WARN: Failed to post msg recv stat: %s", err) + recordRecvUsage(h, append(acceptedTo, acceptedAddTo...), timeutil.TimestampNow().Float64()) + return true +} + +// recordRecvUsage records a stored message against each recipient that +// accepted it: its stored size (body plus attachments), at now — the time this +// host accepted it, not the sender's timestamp, so a backdated message still +// counts towards today's limits. A failure is logged: the message is already +// stored. +func recordRecvUsage(h *FMsgHeader, accepted []FMsgAddress, now float64) { + size := storedSize(h) + for i := range accepted { + if err := postMsgStatRecv(&accepted[i], now, size); err != nil { + log.Printf("WARN: Failed to post msg recv stat for %s: %s", accepted[i].ToString(), err) } } - return true } func downloadMessage(c net.Conn, r io.Reader, h *FMsgHeader, skipData bool) error { diff --git a/cmd/fmsgd/id.go b/cmd/fmsgd/id.go index c639086..d4c8ca9 100644 --- a/cmd/fmsgd/id.go +++ b/cmd/fmsgd/id.go @@ -52,15 +52,15 @@ func getAddressDetail(addr *FMsgAddress) (*AddressDetail, error) { return &detail, nil } -func postMsgStatSend(addr *FMsgAddress, timestamp float64, size int) error { +func postMsgStatSend(addr *FMsgAddress, timestamp float64, size int64) error { return postMsgStat(addr, timestamp, size, true) } -func postMsgStatRecv(addr *FMsgAddress, timestamp float64, size int) error { +func postMsgStatRecv(addr *FMsgAddress, timestamp float64, size int64) error { return postMsgStat(addr, timestamp, size, false) } -func postMsgStat(addr *FMsgAddress, timestamp float64, size int, isSending bool) error { +func postMsgStat(addr *FMsgAddress, timestamp float64, size int64, isSending bool) error { var part string if isSending { part = "send" diff --git a/cmd/fmsgd/usage_test.go b/cmd/fmsgd/usage_test.go new file mode 100644 index 0000000..623758d --- /dev/null +++ b/cmd/fmsgd/usage_test.go @@ -0,0 +1,130 @@ +package main + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" +) + +func TestStoredSize(t *testing.T) { + tests := []struct { + name string + h FMsgHeader + want int64 + }{ + {name: "body only", h: FMsgHeader{Size: 100}, want: 100}, + {name: "compressed body counts expanded", h: FMsgHeader{Flags: FlagDeflate, Size: 40, ExpandedSize: 400}, want: 400}, + { + name: "attachments included", + h: FMsgHeader{Size: 100, Attachments: []FMsgAttachmentHeader{ + {Size: 1000}, + {Flags: 1 << 1, Size: 50, ExpandedSize: 5000}, + }}, + want: 100 + 1000 + 5000, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := storedSize(&tt.h); got != tt.want { + t.Fatalf("storedSize = %d, want %d", got, tt.want) + } + }) + } +} + +func TestRecvLimitExceeded(t *testing.T) { + unlimited := AddressDetail{ + LimitRecvSizeTotal: -1, LimitRecvSizePer1d: -1, LimitRecvCountPer1d: -1, + RecvSizeTotal: 1 << 40, RecvSizePer1d: 1 << 40, RecvCountPer1d: 1 << 40, + } + tests := []struct { + name string + detail AddressDetail + size int64 + want string // substring of the reason, "" when within limits + }{ + {name: "unlimited", detail: unlimited, size: 1 << 30}, + {name: "count fits", detail: AddressDetail{LimitRecvCountPer1d: 10, RecvCountPer1d: 9, LimitRecvSizePer1d: -1, LimitRecvSizeTotal: -1}, size: 1}, + {name: "count full", detail: AddressDetail{LimitRecvCountPer1d: 10, RecvCountPer1d: 10, LimitRecvSizePer1d: -1, LimitRecvSizeTotal: -1}, size: 1, want: "RecvCountPer1d"}, + {name: "day size fits exactly", detail: AddressDetail{LimitRecvCountPer1d: -1, LimitRecvSizePer1d: 100, RecvSizePer1d: 60, LimitRecvSizeTotal: -1}, size: 40}, + {name: "day size over", detail: AddressDetail{LimitRecvCountPer1d: -1, LimitRecvSizePer1d: 100, RecvSizePer1d: 60, LimitRecvSizeTotal: -1}, size: 41, want: "RecvSizePer1d"}, + {name: "total over", detail: AddressDetail{LimitRecvCountPer1d: -1, LimitRecvSizePer1d: -1, LimitRecvSizeTotal: 1000, RecvSizeTotal: 999}, size: 2, want: "RecvSizeTotal"}, + {name: "zero limit refuses", detail: AddressDetail{LimitRecvCountPer1d: 0, LimitRecvSizePer1d: -1, LimitRecvSizeTotal: -1}, size: 0, want: "RecvCountPer1d"}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := recvLimitExceeded(&tt.detail, tt.size) + if tt.want == "" && got != "" { + t.Fatalf("got %q, want within limits", got) + } + if tt.want != "" && !strings.Contains(got, tt.want) { + t.Fatalf("got %q, want %q", got, tt.want) + } + }) + } +} + +// TestRecvLimitIncludesAttachments shows a message whose body fits but whose +// attachments take it over the limit is refused: limits are checked against +// the stored size, the same size that is recorded. +func TestRecvLimitIncludesAttachments(t *testing.T) { + h := FMsgHeader{Size: 10, Attachments: []FMsgAttachmentHeader{{Size: 500}}} + detail := AddressDetail{LimitRecvCountPer1d: -1, LimitRecvSizePer1d: 100, LimitRecvSizeTotal: -1} + if recvLimitExceeded(&detail, int64(h.Size)) != "" { + t.Fatal("body alone should fit") + } + if recvLimitExceeded(&detail, storedSize(&h)) == "" { + t.Fatal("body plus attachments should exceed the daily size limit") + } +} + +func TestRecordRecvUsage(t *testing.T) { + type tx struct { + Address string `json:"address"` + Ts float64 `json:"ts"` + Size int64 `json:"size"` + } + var mu sync.Mutex + var got []tx + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost || r.URL.Path != "/fmsgid/recv" { + t.Errorf("unexpected %s %s", r.Method, r.URL.Path) + w.WriteHeader(http.StatusNotFound) + return + } + var v tx + if err := json.NewDecoder(r.Body).Decode(&v); err != nil { + t.Errorf("decode: %v", err) + } + mu.Lock() + got = append(got, v) + mu.Unlock() + })) + defer srv.Close() + origURI := IDURI + IDURI = srv.URL + defer func() { IDURI = origURI }() + + // The sender's timestamp is a day old; usage must use the accept time. + h := FMsgHeader{ + Timestamp: 1_000, + Flags: FlagDeflate, + Size: 20, + ExpandedSize: 200, + Attachments: []FMsgAttachmentHeader{{Size: 30}}, + } + accepted := []FMsgAddress{{User: "alice", Domain: "example.com"}, {User: "bob", Domain: "example.com"}} + recordRecvUsage(&h, accepted, 90_000) + + if len(got) != 2 { + t.Fatalf("posted %d recv stats, want 2", len(got)) + } + for i, v := range got { + if v.Address != accepted[i].ToString() || v.Ts != 90_000 || v.Size != 230 { + t.Fatalf("stat %d = %+v, want %s at 90000 size 230", i, v, accepted[i].ToString()) + } + } +}