refactor: extract shared mixins and helpers, drop dead abstractions
Introduces shared seams for paginated views, D-pad reorder, media control routing, async singletons and the device method channel, then points the open-coded copies at them. Also removes unused models and duplicated provider/server plumbing, folds the twice-implemented artifact store in the server, and factors the repeated Flutter toolchain prologue in CI into a composite action.
This commit is contained in:
@@ -0,0 +1,434 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"crypto/rand"
|
||||
"errors"
|
||||
"io/fs"
|
||||
"log"
|
||||
"math/big"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"sync"
|
||||
"syscall"
|
||||
"time"
|
||||
)
|
||||
|
||||
// --- Artifact store ---
|
||||
//
|
||||
// Uploaded logs and posters are the same on-disk artifact store: a flat
|
||||
// directory of `<id><ext>` files whose mtime carries the creation time, written
|
||||
// through a temp file, capped by a quota and swept for expiry. Files that
|
||||
// cannot be deleted become pending debt so a failed removal never silently
|
||||
// frees quota. artifactStore implements all of that once; the two flavours
|
||||
// differ only in the policy fields below.
|
||||
|
||||
type artifactRemovalError struct {
|
||||
err error
|
||||
}
|
||||
|
||||
func (e *artifactRemovalError) Error() string {
|
||||
return "artifact removal failed"
|
||||
}
|
||||
|
||||
func (e *artifactRemovalError) Unwrap() error {
|
||||
return e.err
|
||||
}
|
||||
|
||||
var errArtifactOutsideStore = errors.New("artifact path outside store")
|
||||
|
||||
func classifyRemovalError(err error) error {
|
||||
if err == nil || errors.Is(err, fs.ErrNotExist) {
|
||||
return nil
|
||||
}
|
||||
return err
|
||||
}
|
||||
|
||||
func removeArtifact(removeFile func(string) error, root, path string) error {
|
||||
err := classifyRemovalError(removeFile(path))
|
||||
if err == nil {
|
||||
return nil
|
||||
}
|
||||
if !errors.Is(err, syscall.ENOTEMPTY) && !errors.Is(err, syscall.EEXIST) {
|
||||
return &artifactRemovalError{err: err}
|
||||
}
|
||||
if err := removeConfinedDirectory(root, path); err != nil {
|
||||
return &artifactRemovalError{err: err}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func removeConfinedDirectory(root, path string) error {
|
||||
info, err := os.Lstat(path)
|
||||
if err != nil {
|
||||
return classifyRemovalError(err)
|
||||
}
|
||||
if !info.IsDir() {
|
||||
return syscall.ENOTDIR
|
||||
}
|
||||
|
||||
rootPath, err := filepath.Abs(root)
|
||||
if err != nil {
|
||||
return errArtifactOutsideStore
|
||||
}
|
||||
rootPath, err = filepath.EvalSymlinks(rootPath)
|
||||
if err != nil {
|
||||
return errArtifactOutsideStore
|
||||
}
|
||||
artifactPath, err := filepath.Abs(path)
|
||||
if err != nil {
|
||||
return errArtifactOutsideStore
|
||||
}
|
||||
artifactPath, err = filepath.EvalSymlinks(artifactPath)
|
||||
if err != nil {
|
||||
return errArtifactOutsideStore
|
||||
}
|
||||
relative, err := filepath.Rel(rootPath, artifactPath)
|
||||
if err != nil ||
|
||||
relative == "." ||
|
||||
relative == ".." ||
|
||||
filepath.IsAbs(relative) ||
|
||||
strings.HasPrefix(relative, ".."+string(filepath.Separator)) {
|
||||
return errArtifactOutsideStore
|
||||
}
|
||||
return os.RemoveAll(artifactPath)
|
||||
}
|
||||
|
||||
const idChars = "abcdefghijklmnopqrstuvwxyz0123456789"
|
||||
|
||||
func generateID(length int) string {
|
||||
b := make([]byte, length)
|
||||
for i := range b {
|
||||
n, _ := rand.Int(rand.Reader, big.NewInt(int64(len(idChars))))
|
||||
b[i] = idChars[n.Int64()]
|
||||
}
|
||||
return string(b)
|
||||
}
|
||||
|
||||
func validID(id string, length int) bool {
|
||||
if len(id) != length {
|
||||
return false
|
||||
}
|
||||
for _, ch := range id {
|
||||
if !strings.ContainsRune(idChars, ch) {
|
||||
return false
|
||||
}
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
// pendingRemoval is an artifact file that could not be deleted yet. Its size is
|
||||
// unknown when the file could not be stat'ed or is not a regular file.
|
||||
type pendingRemoval struct {
|
||||
size int64
|
||||
sizeKnown bool
|
||||
}
|
||||
|
||||
type artifactEntry struct {
|
||||
Filename string
|
||||
Size int64
|
||||
ContentType string
|
||||
CreatedAt time.Time
|
||||
ExpiresAt time.Time
|
||||
}
|
||||
|
||||
type artifactStore struct {
|
||||
entries map[string]artifactEntry
|
||||
pendingRemovals map[string]pendingRemoval
|
||||
dir string
|
||||
name string // log prefix, e.g. "logs"
|
||||
maxAge time.Duration
|
||||
removeFile func(string) error
|
||||
|
||||
// Policy. generateID is a field so tests can force ID collisions.
|
||||
generateID func() string
|
||||
idFromFilename func(filename string) (string, bool)
|
||||
// acceptLoaded reports whether a file found on disk is a usable artifact
|
||||
// and returns the content type recorded for it.
|
||||
acceptLoaded func(filename string, size int64) (string, bool)
|
||||
// limit caps accountedLocked, measured in the units cost returns:
|
||||
// one per artifact for logs, bytes for posters.
|
||||
limit int64
|
||||
cost func(size int64) int64
|
||||
pendingCost func(pending pendingRemoval) int64
|
||||
// evictToFit admits a new artifact by evicting the oldest live ones;
|
||||
// stores that leave it false reject the upload with errFull instead.
|
||||
evictToFit bool
|
||||
// retryKnownDebtOnPut retries only debt whose size is accounted, leaving
|
||||
// unknown debt to periodic cleanup.
|
||||
retryKnownDebtOnPut bool
|
||||
errFull error
|
||||
|
||||
used int64 // accounted cost of live entries
|
||||
pendingDebt int64 // accounted cost of pending removals
|
||||
unknownPending int // pending removals kept out of the quota
|
||||
startupErr error
|
||||
mu sync.RWMutex
|
||||
}
|
||||
|
||||
func (as *artifactStore) filePath(filename string) string {
|
||||
return filepath.Join(as.dir, filename)
|
||||
}
|
||||
|
||||
func (as *artifactStore) accountedLocked() int64 {
|
||||
return as.used + as.pendingDebt
|
||||
}
|
||||
|
||||
func (as *artifactStore) loadExisting(now time.Time) error {
|
||||
as.mu.Lock()
|
||||
defer as.mu.Unlock()
|
||||
|
||||
files, err := os.ReadDir(as.dir)
|
||||
if err != nil {
|
||||
log.Printf("%s: failed to read dir %s: %v", as.name, as.dir, err)
|
||||
return nil
|
||||
}
|
||||
var removalErr error
|
||||
drop := func(file fs.DirEntry) {
|
||||
size, sizeKnown := dirEntrySize(file)
|
||||
removalErr = errors.Join(removalErr, as.removeUntrackedLocked(file.Name(), size, sizeKnown))
|
||||
}
|
||||
for _, file := range files {
|
||||
filename := file.Name()
|
||||
if file.IsDir() || strings.HasSuffix(filename, ".tmp") {
|
||||
drop(file)
|
||||
continue
|
||||
}
|
||||
id, ok := as.idFromFilename(filename)
|
||||
if !ok {
|
||||
drop(file)
|
||||
continue
|
||||
}
|
||||
info, infoErr := file.Info()
|
||||
if infoErr != nil || !info.Mode().IsRegular() {
|
||||
drop(file)
|
||||
continue
|
||||
}
|
||||
contentType, ok := as.acceptLoaded(filename, info.Size())
|
||||
if !ok {
|
||||
drop(file)
|
||||
continue
|
||||
}
|
||||
// Several extensions can map to one id; keep the first and drop the rest.
|
||||
if _, duplicate := as.entries[id]; duplicate {
|
||||
drop(file)
|
||||
continue
|
||||
}
|
||||
createdAt := info.ModTime()
|
||||
as.entries[id] = artifactEntry{
|
||||
Filename: filename,
|
||||
Size: info.Size(),
|
||||
ContentType: contentType,
|
||||
CreatedAt: createdAt,
|
||||
ExpiresAt: createdAt.Add(as.maxAge),
|
||||
}
|
||||
as.used += as.cost(info.Size())
|
||||
}
|
||||
removalErr = errors.Join(removalErr, as.cleanupExpiredLocked(now))
|
||||
removalErr = errors.Join(removalErr, as.evictOldestLocked(0))
|
||||
return removalErr
|
||||
}
|
||||
|
||||
// put writes data as `<id><ext>` once the quota allows it.
|
||||
func (as *artifactStore) put(data []byte, ext, contentType string, now time.Time) (string, artifactEntry, error) {
|
||||
as.mu.Lock()
|
||||
defer as.mu.Unlock()
|
||||
|
||||
size := int64(len(data))
|
||||
cost := as.cost(size)
|
||||
// Reclaim what the quota can get back without touching live entries.
|
||||
// Removal failures stay accounted as debt instead of blocking the write.
|
||||
_ = as.retryPendingLocked(as.retryKnownDebtOnPut)
|
||||
_ = as.cleanupExpiredLocked(now)
|
||||
var headroom int64
|
||||
if as.evictToFit {
|
||||
headroom = cost
|
||||
}
|
||||
if err := as.evictOldestLocked(headroom); err != nil {
|
||||
return "", artifactEntry{}, err
|
||||
}
|
||||
if as.accountedLocked()+cost > as.limit {
|
||||
return "", artifactEntry{}, as.errFull
|
||||
}
|
||||
|
||||
id := as.generateID()
|
||||
for {
|
||||
if _, exists := as.entries[id]; !exists {
|
||||
if _, err := os.Stat(as.filePath(id + ext)); errors.Is(err, fs.ErrNotExist) {
|
||||
break
|
||||
}
|
||||
}
|
||||
id = as.generateID()
|
||||
}
|
||||
|
||||
filename := id + ext
|
||||
path := as.filePath(filename)
|
||||
tmpPath := path + ".tmp"
|
||||
if err := os.WriteFile(tmpPath, data, 0644); err != nil {
|
||||
as.cleanupFailedTempLocked(tmpPath)
|
||||
return "", artifactEntry{}, err
|
||||
}
|
||||
if err := os.Rename(tmpPath, path); err != nil {
|
||||
as.cleanupFailedTempLocked(tmpPath)
|
||||
return "", artifactEntry{}, err
|
||||
}
|
||||
_ = os.Chtimes(path, now, now)
|
||||
|
||||
entry := artifactEntry{
|
||||
Filename: filename,
|
||||
Size: size,
|
||||
ContentType: contentType,
|
||||
CreatedAt: now,
|
||||
ExpiresAt: now.Add(as.maxAge),
|
||||
}
|
||||
as.entries[id] = entry
|
||||
as.used += cost
|
||||
return id, entry, nil
|
||||
}
|
||||
|
||||
// lookupEntry returns the live entry for id, dropping it when it has expired.
|
||||
// match, when set, rejects entries the caller did not ask for before expiry is
|
||||
// considered, so a mismatched request never triggers a removal.
|
||||
func (as *artifactStore) lookupEntry(
|
||||
id string,
|
||||
now time.Time,
|
||||
match func(artifactEntry) bool,
|
||||
) (artifactEntry, bool, error) {
|
||||
as.mu.Lock()
|
||||
defer as.mu.Unlock()
|
||||
entry, ok := as.entries[id]
|
||||
if !ok || (match != nil && !match(entry)) {
|
||||
return artifactEntry{}, false, nil
|
||||
}
|
||||
if !now.Before(entry.ExpiresAt) {
|
||||
if err := as.deleteEntryLocked(id); err != nil {
|
||||
return artifactEntry{}, false, err
|
||||
}
|
||||
return artifactEntry{}, false, nil
|
||||
}
|
||||
return entry, true, nil
|
||||
}
|
||||
|
||||
func (as *artifactStore) cleanup(now time.Time) error {
|
||||
as.mu.Lock()
|
||||
defer as.mu.Unlock()
|
||||
return as.cleanupLocked(now)
|
||||
}
|
||||
|
||||
func (as *artifactStore) cleanupLocked(now time.Time) error {
|
||||
removalErr := as.retryPendingLocked(false)
|
||||
removalErr = errors.Join(removalErr, as.cleanupExpiredLocked(now))
|
||||
return errors.Join(removalErr, as.evictOldestLocked(0))
|
||||
}
|
||||
|
||||
func (as *artifactStore) cleanupExpiredLocked(now time.Time) error {
|
||||
var removalErr error
|
||||
for id, entry := range as.entries {
|
||||
if !now.Before(entry.ExpiresAt) {
|
||||
removalErr = errors.Join(removalErr, as.deleteEntryLocked(id))
|
||||
}
|
||||
}
|
||||
return removalErr
|
||||
}
|
||||
|
||||
// evictOldestLocked deletes oldest-first until headroom more cost units fit.
|
||||
func (as *artifactStore) evictOldestLocked(headroom int64) error {
|
||||
for as.accountedLocked()+headroom > as.limit && len(as.entries) > 0 {
|
||||
var oldestID string
|
||||
var oldest artifactEntry
|
||||
for id, entry := range as.entries {
|
||||
if oldestID == "" || entry.CreatedAt.Before(oldest.CreatedAt) {
|
||||
oldestID = id
|
||||
oldest = entry
|
||||
}
|
||||
}
|
||||
if err := as.deleteEntryLocked(oldestID); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (as *artifactStore) deleteEntryLocked(id string) error {
|
||||
entry, ok := as.entries[id]
|
||||
if !ok {
|
||||
return nil
|
||||
}
|
||||
if err := removeArtifact(as.removeFile, as.dir, as.filePath(entry.Filename)); err != nil {
|
||||
return err
|
||||
}
|
||||
delete(as.entries, id)
|
||||
as.used -= as.cost(entry.Size)
|
||||
return nil
|
||||
}
|
||||
|
||||
// removeUntrackedLocked deletes a file the index does not own, recording it as
|
||||
// pending debt when the removal fails.
|
||||
func (as *artifactStore) removeUntrackedLocked(filename string, size int64, sizeKnown bool) error {
|
||||
if err := removeArtifact(as.removeFile, as.dir, as.filePath(filename)); err != nil {
|
||||
as.addPendingLocked(filename, size, sizeKnown)
|
||||
return err
|
||||
}
|
||||
as.dropPendingLocked(filename)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (as *artifactStore) retryPendingLocked(knownDebtOnly bool) error {
|
||||
var removalErr error
|
||||
for filename, pending := range as.pendingRemovals {
|
||||
if knownDebtOnly && !pending.sizeKnown {
|
||||
continue
|
||||
}
|
||||
if err := removeArtifact(as.removeFile, as.dir, as.filePath(filename)); err != nil {
|
||||
removalErr = errors.Join(removalErr, err)
|
||||
continue
|
||||
}
|
||||
as.dropPendingLocked(filename)
|
||||
}
|
||||
return removalErr
|
||||
}
|
||||
|
||||
func (as *artifactStore) cleanupFailedTempLocked(tmpPath string) {
|
||||
size, sizeKnown := fileSize(tmpPath)
|
||||
_ = as.removeUntrackedLocked(filepath.Base(tmpPath), size, sizeKnown)
|
||||
}
|
||||
|
||||
func (as *artifactStore) addPendingLocked(filename string, size int64, sizeKnown bool) {
|
||||
if _, exists := as.pendingRemovals[filename]; exists {
|
||||
return
|
||||
}
|
||||
pending := pendingRemoval{size: size, sizeKnown: sizeKnown}
|
||||
as.pendingRemovals[filename] = pending
|
||||
as.pendingDebt += as.pendingCost(pending)
|
||||
if !sizeKnown {
|
||||
as.unknownPending++
|
||||
}
|
||||
}
|
||||
|
||||
func (as *artifactStore) dropPendingLocked(filename string) {
|
||||
pending, exists := as.pendingRemovals[filename]
|
||||
if !exists {
|
||||
return
|
||||
}
|
||||
delete(as.pendingRemovals, filename)
|
||||
as.pendingDebt -= as.pendingCost(pending)
|
||||
if !pending.sizeKnown {
|
||||
as.unknownPending--
|
||||
}
|
||||
}
|
||||
|
||||
func dirEntrySize(file fs.DirEntry) (int64, bool) {
|
||||
info, err := file.Info()
|
||||
if err != nil || !info.Mode().IsRegular() {
|
||||
return 0, false
|
||||
}
|
||||
return info.Size(), true
|
||||
}
|
||||
|
||||
func fileSize(path string) (int64, bool) {
|
||||
info, err := os.Stat(path)
|
||||
if err != nil || !info.Mode().IsRegular() {
|
||||
return 0, false
|
||||
}
|
||||
return info.Size(), true
|
||||
}
|
||||
+72
-570
@@ -13,7 +13,6 @@ import (
|
||||
"io"
|
||||
"io/fs"
|
||||
"log"
|
||||
"math/big"
|
||||
"net"
|
||||
"net/http"
|
||||
"os"
|
||||
@@ -383,100 +382,17 @@ func (r *Room) sendFrom(senderID string, sender *Client, targetID string, msg se
|
||||
}
|
||||
|
||||
// --- Log store ---
|
||||
type artifactRemovalError struct {
|
||||
err error
|
||||
}
|
||||
|
||||
func (e *artifactRemovalError) Error() string {
|
||||
return "artifact removal failed"
|
||||
}
|
||||
|
||||
func (e *artifactRemovalError) Unwrap() error {
|
||||
return e.err
|
||||
}
|
||||
|
||||
var errArtifactOutsideStore = errors.New("artifact path outside store")
|
||||
|
||||
func classifyRemovalError(err error) error {
|
||||
if err == nil || errors.Is(err, fs.ErrNotExist) {
|
||||
return nil
|
||||
}
|
||||
return err
|
||||
}
|
||||
|
||||
func removeArtifact(removeFile func(string) error, root, path string) error {
|
||||
err := classifyRemovalError(removeFile(path))
|
||||
if err == nil {
|
||||
return nil
|
||||
}
|
||||
if !errors.Is(err, syscall.ENOTEMPTY) && !errors.Is(err, syscall.EEXIST) {
|
||||
return &artifactRemovalError{err: err}
|
||||
}
|
||||
if err := removeConfinedDirectory(root, path); err != nil {
|
||||
return &artifactRemovalError{err: err}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func removeConfinedDirectory(root, path string) error {
|
||||
info, err := os.Lstat(path)
|
||||
if err != nil {
|
||||
return classifyRemovalError(err)
|
||||
}
|
||||
if !info.IsDir() {
|
||||
return syscall.ENOTDIR
|
||||
}
|
||||
|
||||
rootPath, err := filepath.Abs(root)
|
||||
if err != nil {
|
||||
return errArtifactOutsideStore
|
||||
}
|
||||
rootPath, err = filepath.EvalSymlinks(rootPath)
|
||||
if err != nil {
|
||||
return errArtifactOutsideStore
|
||||
}
|
||||
artifactPath, err := filepath.Abs(path)
|
||||
if err != nil {
|
||||
return errArtifactOutsideStore
|
||||
}
|
||||
artifactPath, err = filepath.EvalSymlinks(artifactPath)
|
||||
if err != nil {
|
||||
return errArtifactOutsideStore
|
||||
}
|
||||
relative, err := filepath.Rel(rootPath, artifactPath)
|
||||
if err != nil ||
|
||||
relative == "." ||
|
||||
relative == ".." ||
|
||||
filepath.IsAbs(relative) ||
|
||||
strings.HasPrefix(relative, ".."+string(filepath.Separator)) {
|
||||
return errArtifactOutsideStore
|
||||
}
|
||||
return os.RemoveAll(artifactPath)
|
||||
}
|
||||
|
||||
type pendingRemoval struct {
|
||||
size int64
|
||||
sizeKnown bool
|
||||
}
|
||||
|
||||
type logEntry struct {
|
||||
Size int
|
||||
CreatedAt time.Time
|
||||
ExpiresAt time.Time
|
||||
}
|
||||
const logFileExt = ".log"
|
||||
|
||||
var errLogStoreFull = errors.New("log store full")
|
||||
|
||||
// logStore keeps diagnostic uploads capped by artifact count; a full store
|
||||
// rejects new uploads rather than evicting logs someone may still be reading.
|
||||
type logStore struct {
|
||||
entries map[string]logEntry
|
||||
pendingRemovals map[string]pendingRemoval
|
||||
artifactStore
|
||||
rateLimit map[string]time.Time // IP -> last upload time
|
||||
failedLookupRate map[string]*rateLimiter
|
||||
dir string
|
||||
generateID func() string
|
||||
removeFile func(string) error
|
||||
startupErr error
|
||||
mu sync.RWMutex
|
||||
}
|
||||
|
||||
func newLogStore(dir string) *logStore {
|
||||
@@ -488,31 +404,32 @@ func newLogStoreWithRemover(dir string, removeFile func(string) error) *logStore
|
||||
log.Fatalf("failed to create log dir %s: %v", dir, err)
|
||||
}
|
||||
ls := &logStore{
|
||||
entries: make(map[string]logEntry),
|
||||
pendingRemovals: make(map[string]pendingRemoval),
|
||||
artifactStore: artifactStore{
|
||||
entries: make(map[string]artifactEntry),
|
||||
pendingRemovals: make(map[string]pendingRemoval),
|
||||
dir: dir,
|
||||
name: "logs",
|
||||
maxAge: logMaxAge,
|
||||
removeFile: removeFile,
|
||||
generateID: generateLogID,
|
||||
idFromFilename: logIDFromFilename,
|
||||
acceptLoaded: func(_ string, size int64) (string, bool) {
|
||||
return "", size > 0 && size <= maxLogSize
|
||||
},
|
||||
limit: maxLogEntries,
|
||||
cost: func(int64) int64 { return 1 },
|
||||
pendingCost: func(pendingRemoval) int64 { return 1 },
|
||||
errFull: errLogStoreFull,
|
||||
},
|
||||
rateLimit: make(map[string]time.Time),
|
||||
failedLookupRate: make(map[string]*rateLimiter),
|
||||
dir: dir,
|
||||
generateID: generateLogID,
|
||||
removeFile: removeFile,
|
||||
}
|
||||
ls.startupErr = ls.loadExisting(time.Now())
|
||||
return ls
|
||||
}
|
||||
|
||||
func (ls *logStore) filePath(id string) string {
|
||||
return filepath.Join(ls.dir, id+".log")
|
||||
}
|
||||
|
||||
const idChars = "abcdefghijklmnopqrstuvwxyz0123456789"
|
||||
|
||||
func generateID(length int) string {
|
||||
b := make([]byte, length)
|
||||
for i := range b {
|
||||
n, _ := rand.Int(rand.Reader, big.NewInt(int64(len(idChars))))
|
||||
b[i] = idChars[n.Int64()]
|
||||
}
|
||||
return string(b)
|
||||
return ls.artifactStore.filePath(id + logFileExt)
|
||||
}
|
||||
|
||||
func generateLogID() string {
|
||||
@@ -520,149 +437,28 @@ func generateLogID() string {
|
||||
}
|
||||
|
||||
func logIDFromFilename(filename string) (string, bool) {
|
||||
if filepath.Ext(filename) != ".log" {
|
||||
if filepath.Ext(filename) != logFileExt {
|
||||
return "", false
|
||||
}
|
||||
id := strings.TrimSuffix(filename, ".log")
|
||||
id := strings.TrimSuffix(filename, logFileExt)
|
||||
return id, validID(id, logIDLength)
|
||||
}
|
||||
|
||||
func (ls *logStore) loadExisting(now time.Time) error {
|
||||
ls.mu.Lock()
|
||||
defer ls.mu.Unlock()
|
||||
|
||||
files, err := os.ReadDir(ls.dir)
|
||||
if err != nil {
|
||||
log.Printf("logs: failed to read dir %s: %v", ls.dir, err)
|
||||
return nil
|
||||
}
|
||||
var removalErr error
|
||||
for _, file := range files {
|
||||
filename := file.Name()
|
||||
if file.IsDir() || strings.HasSuffix(filename, ".tmp") {
|
||||
removalErr = errors.Join(removalErr, ls.removeUntrackedLocked(filename))
|
||||
continue
|
||||
}
|
||||
id, ok := logIDFromFilename(filename)
|
||||
if !ok {
|
||||
removalErr = errors.Join(removalErr, ls.removeUntrackedLocked(filename))
|
||||
continue
|
||||
}
|
||||
info, infoErr := file.Info()
|
||||
if infoErr != nil || !info.Mode().IsRegular() || info.Size() <= 0 || info.Size() > maxLogSize {
|
||||
removalErr = errors.Join(removalErr, ls.removeUntrackedLocked(filename))
|
||||
continue
|
||||
}
|
||||
createdAt := info.ModTime()
|
||||
ls.entries[id] = logEntry{
|
||||
Size: int(info.Size()),
|
||||
CreatedAt: createdAt,
|
||||
ExpiresAt: createdAt.Add(logMaxAge),
|
||||
}
|
||||
}
|
||||
removalErr = errors.Join(removalErr, ls.cleanupExpiredLocked(now))
|
||||
removalErr = errors.Join(removalErr, ls.evictOldestLocked(maxLogEntries))
|
||||
return removalErr
|
||||
}
|
||||
|
||||
func (ls *logStore) removeUntrackedLocked(filename string) error {
|
||||
if err := removeArtifact(ls.removeFile, ls.dir, filepath.Join(ls.dir, filename)); err != nil {
|
||||
if _, exists := ls.pendingRemovals[filename]; !exists {
|
||||
ls.pendingRemovals[filename] = pendingRemoval{}
|
||||
}
|
||||
return err
|
||||
}
|
||||
delete(ls.pendingRemovals, filename)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (ls *logStore) retryPendingLocked() error {
|
||||
var removalErr error
|
||||
for filename := range ls.pendingRemovals {
|
||||
if err := removeArtifact(ls.removeFile, ls.dir, filepath.Join(ls.dir, filename)); err != nil {
|
||||
removalErr = errors.Join(removalErr, err)
|
||||
continue
|
||||
}
|
||||
delete(ls.pendingRemovals, filename)
|
||||
}
|
||||
return removalErr
|
||||
}
|
||||
|
||||
func (ls *logStore) cleanupFailedTempLocked(tmpPath string) {
|
||||
_ = ls.removeUntrackedLocked(filepath.Base(tmpPath))
|
||||
}
|
||||
|
||||
func (ls *logStore) artifactCountLocked() int {
|
||||
return len(ls.entries) + len(ls.pendingRemovals)
|
||||
}
|
||||
|
||||
func (ls *logStore) store(data []byte, now time.Time) (string, logEntry, error) {
|
||||
func (ls *logStore) store(data []byte, now time.Time) (string, artifactEntry, error) {
|
||||
if len(data) == 0 {
|
||||
return "", logEntry{}, errors.New("empty log")
|
||||
return "", artifactEntry{}, errors.New("empty log")
|
||||
}
|
||||
if len(data) > maxLogSize {
|
||||
return "", logEntry{}, errors.New("log too large")
|
||||
return "", artifactEntry{}, errors.New("log too large")
|
||||
}
|
||||
|
||||
ls.mu.Lock()
|
||||
defer ls.mu.Unlock()
|
||||
_ = ls.retryPendingLocked()
|
||||
_ = ls.cleanupExpiredLocked(now)
|
||||
if err := ls.evictOldestLocked(maxLogEntries); err != nil {
|
||||
return "", logEntry{}, err
|
||||
}
|
||||
if ls.artifactCountLocked() >= maxLogEntries {
|
||||
return "", logEntry{}, errLogStoreFull
|
||||
}
|
||||
|
||||
id := ls.generateID()
|
||||
for {
|
||||
if _, exists := ls.entries[id]; !exists {
|
||||
if _, err := os.Stat(ls.filePath(id)); errors.Is(err, fs.ErrNotExist) {
|
||||
break
|
||||
}
|
||||
}
|
||||
id = ls.generateID()
|
||||
}
|
||||
|
||||
path := ls.filePath(id)
|
||||
tmpPath := path + ".tmp"
|
||||
if err := os.WriteFile(tmpPath, data, 0644); err != nil {
|
||||
ls.cleanupFailedTempLocked(tmpPath)
|
||||
return "", logEntry{}, err
|
||||
}
|
||||
if err := os.Rename(tmpPath, path); err != nil {
|
||||
ls.cleanupFailedTempLocked(tmpPath)
|
||||
return "", logEntry{}, err
|
||||
}
|
||||
_ = os.Chtimes(path, now, now)
|
||||
|
||||
entry := logEntry{
|
||||
Size: len(data),
|
||||
CreatedAt: now,
|
||||
ExpiresAt: now.Add(logMaxAge),
|
||||
}
|
||||
ls.entries[id] = entry
|
||||
return id, entry, nil
|
||||
return ls.put(data, logFileExt, "", now)
|
||||
}
|
||||
|
||||
func (ls *logStore) lookup(id string, now time.Time) (logEntry, bool, error) {
|
||||
func (ls *logStore) lookup(id string, now time.Time) (artifactEntry, bool, error) {
|
||||
if !validID(id, logIDLength) {
|
||||
return logEntry{}, false, nil
|
||||
return artifactEntry{}, false, nil
|
||||
}
|
||||
ls.mu.Lock()
|
||||
defer ls.mu.Unlock()
|
||||
entry, ok := ls.entries[id]
|
||||
if !ok {
|
||||
return logEntry{}, false, nil
|
||||
}
|
||||
if !now.Before(entry.ExpiresAt) {
|
||||
if err := ls.deleteEntryLocked(id); err != nil {
|
||||
return logEntry{}, false, err
|
||||
}
|
||||
return logEntry{}, false, nil
|
||||
}
|
||||
return entry, true, nil
|
||||
return ls.lookupEntry(id, now, nil)
|
||||
}
|
||||
|
||||
func (ls *logStore) allowFailedLookup(source string, now time.Time) bool {
|
||||
@@ -680,53 +476,10 @@ func (ls *logStore) allowFailedLookup(source string, now time.Time) bool {
|
||||
return limiter.allowAt(now)
|
||||
}
|
||||
|
||||
func (ls *logStore) cleanupExpiredLocked(now time.Time) error {
|
||||
var removalErr error
|
||||
for id, entry := range ls.entries {
|
||||
if !now.Before(entry.ExpiresAt) {
|
||||
removalErr = errors.Join(removalErr, ls.deleteEntryLocked(id))
|
||||
}
|
||||
}
|
||||
return removalErr
|
||||
}
|
||||
|
||||
func (ls *logStore) evictOldestLocked(limit int) error {
|
||||
for ls.artifactCountLocked() > limit {
|
||||
var oldestID string
|
||||
var oldest logEntry
|
||||
for id, entry := range ls.entries {
|
||||
if oldestID == "" || entry.CreatedAt.Before(oldest.CreatedAt) {
|
||||
oldestID = id
|
||||
oldest = entry
|
||||
}
|
||||
}
|
||||
if oldestID == "" {
|
||||
return nil
|
||||
}
|
||||
if err := ls.deleteEntryLocked(oldestID); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (ls *logStore) deleteEntryLocked(id string) error {
|
||||
if _, ok := ls.entries[id]; !ok {
|
||||
return nil
|
||||
}
|
||||
if err := removeArtifact(ls.removeFile, ls.dir, ls.filePath(id)); err != nil {
|
||||
return err
|
||||
}
|
||||
delete(ls.entries, id)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (ls *logStore) cleanup(now time.Time) error {
|
||||
ls.mu.Lock()
|
||||
defer ls.mu.Unlock()
|
||||
removalErr := ls.retryPendingLocked()
|
||||
removalErr = errors.Join(removalErr, ls.cleanupExpiredLocked(now))
|
||||
removalErr = errors.Join(removalErr, ls.evictOldestLocked(maxLogEntries))
|
||||
removalErr := ls.cleanupLocked(now)
|
||||
cleanupRateWindows(ls.rateLimit, now, logRateInterval)
|
||||
cleanupRateLimiters(ls.failedLookupRate, now, nil)
|
||||
return removalErr
|
||||
@@ -734,26 +487,12 @@ func (ls *logStore) cleanup(now time.Time) error {
|
||||
|
||||
// --- Poster store ---
|
||||
|
||||
type posterEntry struct {
|
||||
Filename string
|
||||
Size int64
|
||||
ContentType string
|
||||
CreatedAt time.Time
|
||||
ExpiresAt time.Time
|
||||
}
|
||||
var errPosterStoreFull = errors.New("poster store full")
|
||||
|
||||
// posterStore caps shared posters by accounted bytes and evicts the oldest to
|
||||
// admit a new upload.
|
||||
type posterStore struct {
|
||||
entries map[string]posterEntry
|
||||
pendingRemovals map[string]pendingRemoval
|
||||
dir string
|
||||
maxBytes int64
|
||||
maxAge time.Duration
|
||||
totalBytes int64
|
||||
pendingBytes int64
|
||||
unknownPending int
|
||||
removeFile func(string) error
|
||||
startupErr error
|
||||
mu sync.RWMutex
|
||||
artifactStore
|
||||
}
|
||||
|
||||
func newPosterStore(dir string, maxBytes int64, maxAge time.Duration) *posterStore {
|
||||
@@ -769,22 +508,37 @@ func newPosterStoreWithRemover(
|
||||
if err := os.MkdirAll(dir, 0755); err != nil {
|
||||
log.Fatalf("failed to create poster dir %s: %v", dir, err)
|
||||
}
|
||||
ps := &posterStore{
|
||||
entries: make(map[string]posterEntry),
|
||||
ps := &posterStore{artifactStore{
|
||||
entries: make(map[string]artifactEntry),
|
||||
pendingRemovals: make(map[string]pendingRemoval),
|
||||
dir: dir,
|
||||
maxBytes: maxBytes,
|
||||
name: "posters",
|
||||
maxAge: maxAge,
|
||||
removeFile: removeFile,
|
||||
}
|
||||
generateID: generatePosterID,
|
||||
idFromFilename: posterIDFromFilename,
|
||||
acceptLoaded: func(filename string, _ int64) (string, bool) {
|
||||
return posterContentTypeForExt(filepath.Ext(filename))
|
||||
},
|
||||
limit: maxBytes,
|
||||
cost: func(size int64) int64 { return size },
|
||||
pendingCost: func(pending pendingRemoval) int64 {
|
||||
// Unknown debt cannot be sized safely, so it is kept out of the
|
||||
// quota: a permanent directory or stat failure must not deny
|
||||
// otherwise capacity-safe uploads.
|
||||
if !pending.sizeKnown {
|
||||
return 0
|
||||
}
|
||||
return pending.size
|
||||
},
|
||||
evictToFit: true,
|
||||
retryKnownDebtOnPut: true,
|
||||
errFull: errPosterStoreFull,
|
||||
}}
|
||||
ps.startupErr = ps.loadExisting(time.Now())
|
||||
return ps
|
||||
}
|
||||
|
||||
func (ps *posterStore) filePath(filename string) string {
|
||||
return filepath.Join(ps.dir, filename)
|
||||
}
|
||||
|
||||
func generatePosterID() string {
|
||||
return generateID(posterIDLength)
|
||||
}
|
||||
@@ -819,18 +573,6 @@ func posterContentTypeForExt(ext string) (string, bool) {
|
||||
}
|
||||
}
|
||||
|
||||
func validID(id string, length int) bool {
|
||||
if len(id) != length {
|
||||
return false
|
||||
}
|
||||
for _, ch := range id {
|
||||
if !strings.ContainsRune(idChars, ch) {
|
||||
return false
|
||||
}
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
func posterIDFromFilename(filename string) (string, bool) {
|
||||
if filename == "" || strings.ContainsAny(filename, `/\\`) {
|
||||
return "", false
|
||||
@@ -846,269 +588,29 @@ func posterIDFromFilename(filename string) (string, bool) {
|
||||
return id, true
|
||||
}
|
||||
|
||||
func (ps *posterStore) loadExisting(now time.Time) error {
|
||||
ps.mu.Lock()
|
||||
defer ps.mu.Unlock()
|
||||
|
||||
files, err := os.ReadDir(ps.dir)
|
||||
if err != nil {
|
||||
log.Printf("posters: failed to read dir %s: %v", ps.dir, err)
|
||||
return nil
|
||||
}
|
||||
var removalErr error
|
||||
for _, file := range files {
|
||||
filename := file.Name()
|
||||
if file.IsDir() || strings.HasSuffix(filename, ".tmp") {
|
||||
size, known := posterArtifactSize(file)
|
||||
removalErr = errors.Join(
|
||||
removalErr,
|
||||
ps.removeUntrackedLocked(filename, size, known),
|
||||
)
|
||||
continue
|
||||
}
|
||||
id, ok := posterIDFromFilename(filename)
|
||||
if !ok {
|
||||
size, known := posterArtifactSize(file)
|
||||
removalErr = errors.Join(
|
||||
removalErr,
|
||||
ps.removeUntrackedLocked(filename, size, known),
|
||||
)
|
||||
continue
|
||||
}
|
||||
info, infoErr := file.Info()
|
||||
if infoErr != nil || !info.Mode().IsRegular() {
|
||||
removalErr = errors.Join(
|
||||
removalErr,
|
||||
ps.removeUntrackedLocked(filename, 0, false),
|
||||
)
|
||||
continue
|
||||
}
|
||||
if _, duplicate := ps.entries[id]; duplicate {
|
||||
removalErr = errors.Join(
|
||||
removalErr,
|
||||
ps.removeUntrackedLocked(filename, info.Size(), true),
|
||||
)
|
||||
continue
|
||||
}
|
||||
createdAt := info.ModTime()
|
||||
contentType, _ := posterContentTypeForExt(filepath.Ext(filename))
|
||||
entry := posterEntry{
|
||||
Filename: filename,
|
||||
Size: info.Size(),
|
||||
ContentType: contentType,
|
||||
CreatedAt: createdAt,
|
||||
ExpiresAt: createdAt.Add(ps.maxAge),
|
||||
}
|
||||
ps.entries[id] = entry
|
||||
ps.totalBytes += entry.Size
|
||||
}
|
||||
removalErr = errors.Join(removalErr, ps.cleanupExpiredLocked(now))
|
||||
removalErr = errors.Join(removalErr, ps.evictOldestLocked(0))
|
||||
return removalErr
|
||||
}
|
||||
|
||||
func posterArtifactSize(file fs.DirEntry) (int64, bool) {
|
||||
info, err := file.Info()
|
||||
if err != nil || !info.Mode().IsRegular() {
|
||||
return 0, false
|
||||
}
|
||||
return info.Size(), true
|
||||
}
|
||||
|
||||
func (ps *posterStore) addPendingLocked(filename string, size int64, known bool) {
|
||||
if _, exists := ps.pendingRemovals[filename]; exists {
|
||||
return
|
||||
}
|
||||
ps.pendingRemovals[filename] = pendingRemoval{size: size, sizeKnown: known}
|
||||
if known {
|
||||
ps.pendingBytes += size
|
||||
} else {
|
||||
ps.unknownPending++
|
||||
}
|
||||
}
|
||||
|
||||
func (ps *posterStore) removeUntrackedLocked(filename string, size int64, known bool) error {
|
||||
if err := removeArtifact(ps.removeFile, ps.dir, ps.filePath(filename)); err != nil {
|
||||
ps.addPendingLocked(filename, size, known)
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (ps *posterStore) retryPendingLocked(knownOnly bool) error {
|
||||
var removalErr error
|
||||
for filename, pending := range ps.pendingRemovals {
|
||||
if knownOnly && !pending.sizeKnown {
|
||||
continue
|
||||
}
|
||||
if err := removeArtifact(ps.removeFile, ps.dir, ps.filePath(filename)); err != nil {
|
||||
removalErr = errors.Join(removalErr, err)
|
||||
continue
|
||||
}
|
||||
delete(ps.pendingRemovals, filename)
|
||||
if pending.sizeKnown {
|
||||
ps.pendingBytes -= pending.size
|
||||
} else {
|
||||
ps.unknownPending--
|
||||
}
|
||||
}
|
||||
return removalErr
|
||||
}
|
||||
|
||||
func (ps *posterStore) cleanupFailedTempLocked(tmpPath string) {
|
||||
if err := removeArtifact(ps.removeFile, ps.dir, tmpPath); err == nil {
|
||||
return
|
||||
}
|
||||
info, statErr := os.Stat(tmpPath)
|
||||
known := statErr == nil && info.Mode().IsRegular()
|
||||
var size int64
|
||||
if known {
|
||||
size = info.Size()
|
||||
}
|
||||
ps.addPendingLocked(filepath.Base(tmpPath), size, known)
|
||||
}
|
||||
|
||||
func (ps *posterStore) accountedBytesLocked() int64 {
|
||||
return ps.totalBytes + ps.pendingBytes
|
||||
}
|
||||
|
||||
func (ps *posterStore) store(data []byte, contentType string, now time.Time) (string, posterEntry, error) {
|
||||
func (ps *posterStore) store(data []byte, contentType string, now time.Time) (string, artifactEntry, error) {
|
||||
entrySize := int64(len(data))
|
||||
if entrySize <= 0 {
|
||||
return "", posterEntry{}, errors.New("empty poster")
|
||||
return "", artifactEntry{}, errors.New("empty poster")
|
||||
}
|
||||
if entrySize > ps.maxBytes {
|
||||
return "", posterEntry{}, errors.New("poster exceeds store size")
|
||||
if entrySize > ps.limit {
|
||||
return "", artifactEntry{}, errors.New("poster exceeds store size")
|
||||
}
|
||||
ext, ok := posterExtForContentType(contentType)
|
||||
if !ok {
|
||||
return "", posterEntry{}, errors.New("unsupported poster type")
|
||||
return "", artifactEntry{}, errors.New("unsupported poster type")
|
||||
}
|
||||
|
||||
ps.mu.Lock()
|
||||
defer ps.mu.Unlock()
|
||||
|
||||
// Known regular-file debt counts against quota and is retried on demand.
|
||||
// Unknown artifacts are left to periodic cleanup: their size cannot be
|
||||
// accounted safely, and a permanent directory or stat failure must not
|
||||
// deny otherwise capacity-safe uploads.
|
||||
_ = ps.retryPendingLocked(true)
|
||||
_ = ps.cleanupExpiredLocked(now)
|
||||
if err := ps.evictOldestLocked(entrySize); err != nil {
|
||||
return "", posterEntry{}, err
|
||||
}
|
||||
if ps.accountedBytesLocked()+entrySize > ps.maxBytes {
|
||||
return "", posterEntry{}, errors.New("poster store full")
|
||||
}
|
||||
|
||||
id := generatePosterID()
|
||||
for {
|
||||
if _, exists := ps.entries[id]; !exists {
|
||||
if _, err := os.Stat(ps.filePath(id + ext)); errors.Is(err, fs.ErrNotExist) {
|
||||
break
|
||||
}
|
||||
}
|
||||
id = generatePosterID()
|
||||
}
|
||||
|
||||
filename := id + ext
|
||||
path := ps.filePath(filename)
|
||||
tmpPath := path + ".tmp"
|
||||
if err := os.WriteFile(tmpPath, data, 0644); err != nil {
|
||||
ps.cleanupFailedTempLocked(tmpPath)
|
||||
return "", posterEntry{}, err
|
||||
}
|
||||
if err := os.Rename(tmpPath, path); err != nil {
|
||||
ps.cleanupFailedTempLocked(tmpPath)
|
||||
return "", posterEntry{}, err
|
||||
}
|
||||
_ = os.Chtimes(path, now, now)
|
||||
|
||||
entry := posterEntry{
|
||||
Filename: filename,
|
||||
Size: entrySize,
|
||||
ContentType: strings.ToLower(strings.SplitN(contentType, ";", 2)[0]),
|
||||
CreatedAt: now,
|
||||
ExpiresAt: now.Add(ps.maxAge),
|
||||
}
|
||||
ps.entries[id] = entry
|
||||
ps.totalBytes += entry.Size
|
||||
return id, entry, nil
|
||||
return ps.put(data, ext, strings.ToLower(strings.SplitN(contentType, ";", 2)[0]), now)
|
||||
}
|
||||
|
||||
func (ps *posterStore) lookup(filename string, now time.Time) (posterEntry, bool, error) {
|
||||
func (ps *posterStore) lookup(filename string, now time.Time) (artifactEntry, bool, error) {
|
||||
id, ok := posterIDFromFilename(filename)
|
||||
if !ok {
|
||||
return posterEntry{}, false, nil
|
||||
return artifactEntry{}, false, nil
|
||||
}
|
||||
|
||||
ps.mu.Lock()
|
||||
defer ps.mu.Unlock()
|
||||
entry, ok := ps.entries[id]
|
||||
if !ok || entry.Filename != filename {
|
||||
return posterEntry{}, false, nil
|
||||
}
|
||||
if !now.Before(entry.ExpiresAt) {
|
||||
if err := ps.deleteEntryLocked(id); err != nil {
|
||||
return posterEntry{}, false, err
|
||||
}
|
||||
return posterEntry{}, false, nil
|
||||
}
|
||||
return entry, true, nil
|
||||
}
|
||||
|
||||
func (ps *posterStore) cleanup(now time.Time) error {
|
||||
ps.mu.Lock()
|
||||
defer ps.mu.Unlock()
|
||||
removalErr := ps.retryPendingLocked(false)
|
||||
removalErr = errors.Join(removalErr, ps.cleanupExpiredLocked(now))
|
||||
removalErr = errors.Join(removalErr, ps.evictOldestLocked(0))
|
||||
return removalErr
|
||||
}
|
||||
|
||||
func (ps *posterStore) cleanupExpiredLocked(now time.Time) error {
|
||||
var removalErr error
|
||||
for id, entry := range ps.entries {
|
||||
if !now.Before(entry.ExpiresAt) {
|
||||
removalErr = errors.Join(removalErr, ps.deleteEntryLocked(id))
|
||||
}
|
||||
}
|
||||
return removalErr
|
||||
}
|
||||
|
||||
func (ps *posterStore) evictOldestLocked(extraBytes int64) error {
|
||||
for ps.accountedBytesLocked()+extraBytes > ps.maxBytes && len(ps.entries) > 0 {
|
||||
var oldestID string
|
||||
var oldest posterEntry
|
||||
first := true
|
||||
for id, entry := range ps.entries {
|
||||
if first || entry.CreatedAt.Before(oldest.CreatedAt) {
|
||||
oldestID = id
|
||||
oldest = entry
|
||||
first = false
|
||||
}
|
||||
}
|
||||
if oldestID == "" {
|
||||
return nil
|
||||
}
|
||||
if err := ps.deleteEntryLocked(oldestID); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (ps *posterStore) deleteEntryLocked(id string) error {
|
||||
entry, ok := ps.entries[id]
|
||||
if !ok {
|
||||
return nil
|
||||
}
|
||||
if err := removeArtifact(ps.removeFile, ps.dir, ps.filePath(entry.Filename)); err != nil {
|
||||
return err
|
||||
}
|
||||
delete(ps.entries, id)
|
||||
ps.totalBytes -= entry.Size
|
||||
return nil
|
||||
return ps.lookupEntry(id, now, func(entry artifactEntry) bool {
|
||||
return entry.Filename == filename
|
||||
})
|
||||
}
|
||||
|
||||
// --- Snapshotter (single-writer, debounced, atomic disk persistence) ---
|
||||
@@ -1626,7 +1128,7 @@ func (s *Server) handleGetLogs(w http.ResponseWriter, r *http.Request) {
|
||||
}
|
||||
|
||||
type lookupResult struct {
|
||||
entry logEntry
|
||||
entry artifactEntry
|
||||
data []byte
|
||||
status int
|
||||
message string
|
||||
@@ -1688,7 +1190,7 @@ func (s *Server) handleGetLogs(w http.ResponseWriter, r *http.Request) {
|
||||
}
|
||||
|
||||
w.Header().Set("Content-Type", "text/plain; charset=utf-8")
|
||||
w.Header().Set("Content-Length", strconv.Itoa(lookup.entry.Size))
|
||||
w.Header().Set("Content-Length", strconv.FormatInt(lookup.entry.Size, 10))
|
||||
if written, err := w.Write(lookup.data); err != nil || written != len(lookup.data) {
|
||||
log.Printf("logs: response write failed")
|
||||
}
|
||||
|
||||
+24
-24
@@ -4316,7 +4316,7 @@ func snapshotPosterStore(t *testing.T, store *posterStore) (int, int64, []string
|
||||
t.Helper()
|
||||
store.mu.RLock()
|
||||
entryCount := len(store.entries)
|
||||
totalBytes := store.totalBytes
|
||||
totalBytes := store.used
|
||||
store.mu.RUnlock()
|
||||
files, err := os.ReadDir(store.dir)
|
||||
if err != nil {
|
||||
@@ -4701,7 +4701,7 @@ func TestPosterStoreEvictsOldestOverQuota(t *testing.T) {
|
||||
ps.mu.RLock()
|
||||
_, hasFirst := ps.entries[id1]
|
||||
_, hasSecond := ps.entries[id2]
|
||||
total := ps.totalBytes
|
||||
total := ps.used
|
||||
ps.mu.RUnlock()
|
||||
|
||||
if hasFirst {
|
||||
@@ -4711,7 +4711,7 @@ func TestPosterStoreEvictsOldestOverQuota(t *testing.T) {
|
||||
t.Fatal("newest poster should remain")
|
||||
}
|
||||
if total != int64(len(payload)) {
|
||||
t.Fatalf("totalBytes=%d want %d", total, len(payload))
|
||||
t.Fatalf("stored bytes=%d want %d", total, len(payload))
|
||||
}
|
||||
if _, err := os.Stat(ps.filePath(entry1.Filename)); !os.IsNotExist(err) {
|
||||
t.Fatalf("oldest file still exists or stat failed unexpectedly: %v", err)
|
||||
@@ -4735,13 +4735,13 @@ func TestPosterStoreCleanupExpiresOldPosters(t *testing.T) {
|
||||
|
||||
ps.mu.RLock()
|
||||
_, exists := ps.entries[id]
|
||||
total := ps.totalBytes
|
||||
total := ps.used
|
||||
ps.mu.RUnlock()
|
||||
if exists {
|
||||
t.Fatal("expired poster should have been removed")
|
||||
}
|
||||
if total != 0 {
|
||||
t.Fatalf("totalBytes=%d want 0", total)
|
||||
t.Fatalf("stored bytes=%d want 0", total)
|
||||
}
|
||||
if _, err := os.Stat(ps.filePath(entry.Filename)); !os.IsNotExist(err) {
|
||||
t.Fatalf("expired file still exists or stat failed unexpectedly: %v", err)
|
||||
@@ -4790,7 +4790,7 @@ func TestLogStoreRemovalFailureRetainsEntryUntilRetry(t *testing.T) {
|
||||
}
|
||||
ls.mu.RLock()
|
||||
_, indexed := ls.entries[id]
|
||||
artifacts := ls.artifactCountLocked()
|
||||
artifacts := ls.accountedLocked()
|
||||
ls.mu.RUnlock()
|
||||
if !indexed || artifacts != 1 {
|
||||
t.Fatalf("failed removal changed metadata: indexed=%v artifacts=%d", indexed, artifacts)
|
||||
@@ -4927,7 +4927,7 @@ func TestLogStoreTracksFailedTempCleanup(t *testing.T) {
|
||||
}
|
||||
ls.mu.RLock()
|
||||
_, pending := ls.pendingRemovals[filepath.Base(tmpPath)]
|
||||
artifacts := ls.artifactCountLocked()
|
||||
artifacts := ls.accountedLocked()
|
||||
ls.mu.RUnlock()
|
||||
if !pending || artifacts != 1 {
|
||||
t.Fatalf("temp cleanup not tracked: pending=%v artifacts=%d", pending, artifacts)
|
||||
@@ -4974,7 +4974,7 @@ func TestLogStoreStartupReconcilesLiveAndPendingRemovals(t *testing.T) {
|
||||
ls.mu.RLock()
|
||||
_, live := ls.entries[expiredID]
|
||||
pending := len(ls.pendingRemovals)
|
||||
artifacts := ls.artifactCountLocked()
|
||||
artifacts := ls.accountedLocked()
|
||||
ls.mu.RUnlock()
|
||||
if !live || pending != 2 || artifacts != 3 {
|
||||
t.Fatalf("startup accounting: live=%v pending=%d artifacts=%d", live, pending, artifacts)
|
||||
@@ -4992,7 +4992,7 @@ func TestLogStoreStartupReconcilesLiveAndPendingRemovals(t *testing.T) {
|
||||
}
|
||||
restarted := newLogStore(dir)
|
||||
restarted.mu.RLock()
|
||||
restartedArtifacts := restarted.artifactCountLocked()
|
||||
restartedArtifacts := restarted.accountedLocked()
|
||||
_, newLogRestored := restarted.entries[newID]
|
||||
restarted.mu.RUnlock()
|
||||
if restartedArtifacts != 1 || !newLogRestored {
|
||||
@@ -5084,14 +5084,14 @@ func TestPosterQuotaRemovalFailureDoesNotReclaimAccounting(t *testing.T) {
|
||||
if !errors.Is(err, fs.ErrPermission) {
|
||||
t.Fatalf("quota store error=%v want permission error", err)
|
||||
}
|
||||
if newID != "" || newEntry != (posterEntry{}) {
|
||||
if newID != "" || newEntry != (artifactEntry{}) {
|
||||
t.Fatalf("failed store returned success values: id=%q entry=%+v", newID, newEntry)
|
||||
}
|
||||
ps.mu.RLock()
|
||||
_, retained := ps.entries[oldID]
|
||||
total := ps.totalBytes
|
||||
pending := ps.pendingBytes
|
||||
accounted := ps.accountedBytesLocked()
|
||||
total := ps.used
|
||||
pending := ps.pendingDebt
|
||||
accounted := ps.accountedLocked()
|
||||
ps.mu.RUnlock()
|
||||
if !retained || total != int64(len(payload)) || pending != 0 {
|
||||
t.Fatalf("failed eviction accounting: retained=%v total=%d pending=%d", retained, total, pending)
|
||||
@@ -5112,8 +5112,8 @@ func TestPosterQuotaRemovalFailureDoesNotReclaimAccounting(t *testing.T) {
|
||||
t.Fatalf("old poster remove calls=%d want 2", remover.callCount(oldPath))
|
||||
}
|
||||
ps.mu.RLock()
|
||||
accounted = ps.accountedBytesLocked()
|
||||
total = ps.totalBytes
|
||||
accounted = ps.accountedLocked()
|
||||
total = ps.used
|
||||
ps.mu.RUnlock()
|
||||
if total != int64(len(payload)) || regularFileBytes(t, dir) != accounted {
|
||||
t.Fatalf("retry accounting: total=%d accounted=%d physical=%d", total, accounted, regularFileBytes(t, dir))
|
||||
@@ -5142,7 +5142,7 @@ func TestPosterExpiredRemovalFailureAndErrNotExistAreExactOnce(t *testing.T) {
|
||||
}
|
||||
ps.mu.RLock()
|
||||
_, retained := ps.entries[id]
|
||||
total := ps.totalBytes
|
||||
total := ps.used
|
||||
ps.mu.RUnlock()
|
||||
if !retained || total != entry.Size {
|
||||
t.Fatalf("failed expiry accounting: retained=%v total=%d", retained, total)
|
||||
@@ -5159,10 +5159,10 @@ func TestPosterExpiredRemovalFailureAndErrNotExistAreExactOnce(t *testing.T) {
|
||||
t.Fatalf("remove calls=%d want 2", remover.callCount(path))
|
||||
}
|
||||
ps.mu.RLock()
|
||||
total = ps.totalBytes
|
||||
total = ps.used
|
||||
ps.mu.RUnlock()
|
||||
if total != 0 {
|
||||
t.Fatalf("totalBytes=%d want 0", total)
|
||||
t.Fatalf("stored bytes=%d want 0", total)
|
||||
}
|
||||
})
|
||||
|
||||
@@ -5194,10 +5194,10 @@ func TestPosterExpiredRemovalFailureAndErrNotExistAreExactOnce(t *testing.T) {
|
||||
t.Fatalf("remove calls=%d want 1", remover.callCount(path))
|
||||
}
|
||||
ps.mu.RLock()
|
||||
total := ps.totalBytes
|
||||
total := ps.used
|
||||
ps.mu.RUnlock()
|
||||
if total != 0 {
|
||||
t.Fatalf("totalBytes=%d want 0", total)
|
||||
t.Fatalf("stored bytes=%d want 0", total)
|
||||
}
|
||||
})
|
||||
}
|
||||
@@ -5219,8 +5219,8 @@ func TestPosterStoreKnownCleanupDebtConsumesCapacityAndRetries(t *testing.T) {
|
||||
t.Fatal("upload exceeded capacity after known stale bytes were accounted")
|
||||
}
|
||||
ps.mu.RLock()
|
||||
pendingBytes := ps.pendingBytes
|
||||
accountedBytes := ps.accountedBytesLocked()
|
||||
pendingBytes := ps.pendingDebt
|
||||
accountedBytes := ps.accountedLocked()
|
||||
ps.mu.RUnlock()
|
||||
if pendingBytes != 4 || accountedBytes != 4 {
|
||||
t.Fatalf("known debt accounting: pending=%d accounted=%d, want 4", pendingBytes, accountedBytes)
|
||||
@@ -5236,7 +5236,7 @@ func TestPosterStoreKnownCleanupDebtConsumesCapacityAndRetries(t *testing.T) {
|
||||
t.Fatalf("stored entry size=%d, want 2", entry.Size)
|
||||
}
|
||||
ps.mu.RLock()
|
||||
pendingBytes = ps.pendingBytes
|
||||
pendingBytes = ps.pendingDebt
|
||||
ps.mu.RUnlock()
|
||||
if pendingBytes != 0 {
|
||||
t.Fatalf("known debt remained after successful retry: %d bytes", pendingBytes)
|
||||
@@ -5389,7 +5389,7 @@ func TestPosterHandlerRejectsUploadWhenQuotaRemovalFails(t *testing.T) {
|
||||
}
|
||||
posters.mu.RLock()
|
||||
_, retained := posters.entries[oldID]
|
||||
total := posters.totalBytes
|
||||
total := posters.used
|
||||
posters.mu.RUnlock()
|
||||
if !retained || total != int64(len(payload)) {
|
||||
t.Fatalf("failed upload changed old poster: retained=%v total=%d", retained, total)
|
||||
|
||||
Reference in New Issue
Block a user