fix: admin/monitor: 'file' could be stale on recycled connection
Massimo Melina committed
Mar 12, 2022 at 21:06 UTC
cc95c94296dadc7a1483673accd9dff948bdffa7
4 files changed
+44
-42
server/src/index.ts
+1
-1
@@ -20,7 +20,7 @@ app.use(someSecurity)
20
.use(headRequests)
21
.use(log())
22
.use(pluginsMiddleware())
23
- .use(throttler())
23
+ .use(throttler)
24
.use(gzipper)
25
.use(mount(API_URI, apiMiddleware({ ...frontEndApis, ...adminApis })))
26
.use(serveGuiAndSharedFiles)
server/src/middlewares.ts
+6
-2
@@ -13,9 +13,9 @@ import { serveFileNode } from './serveFile'
13
import { serveGuiFiles } from './serveGuiFiles'
14
import mount from 'koa-mount'
15
import { Readable } from 'stream'
16
-import { getAccount, getCurrentUsername, getCurrentUsernameExpanded } from './perm'
16
+import { getAccount, getCurrentUsername } from './perm'
17
import { getConfig, subscribeConfig } from './config'
18
-import { getConnections } from './connections'
18
+import { getConnections, socket2connection, updateConnection } from './connections'
19
import { Socket } from 'net'
20
21
export const gzipper = compress({
@@ -113,6 +113,10 @@ function applyBlock(socket: Socket) {
113
}
114
115
export const prepareState: Koa.Middleware = async (ctx, next) => {
116
+ // calculate these once and for all
117
ctx.state.account = getAccount(getCurrentUsername(ctx))
118
+ const conn = ctx.state.connection = socket2connection(ctx.socket)
119
+ if (conn?.path) // leftover of connection reused for a new request
120
+ updateConnection(conn, { path: '' })
121
await next()
122
}
server/src/serveFile.ts
+1
-1
@@ -54,7 +54,7 @@ export function serveFile(source:string, mime?:string, modifier?:(s:string)=>str
54
if (ctx.fresh)
55
return ctx.status = 304
56
57
- const conn = socket2connection(ctx.socket)
57
+ const conn = ctx.state.connection
58
if (conn)
59
updateConnection(conn, { path: ctx.path })
60
if (modifier)
server/src/throttler.ts
+36
-38
@@ -22,47 +22,45 @@ const ip2group: Record<string, {
22
const SymThrStr = Symbol('stream')
23
const SymTimeout = Symbol('timeout')
24
25
-export function throttler(): Koa.Middleware {
26
- return async (ctx, next) => {
27
- await next()
28
- const { body } = ctx
29
- if (!body || !(body instanceof Readable) || ctx.state.account?.ignore_limits)
30
- return
31
- const ipGroup = getOrSet(ip2group, ctx.ip, ()=> {
32
- const tg = new ThrottleGroup(Infinity, mainThrottleGroup)
33
- const unsub = subscribeConfig({ k:'max_kbps_per_ip', defaultValue:null }, v =>
34
- tg.updateLimit(v ?? Infinity))
35
- return { group:tg, count:0, destroy: unsub }
36
- })
37
- const conn = socket2connection(ctx.socket)
38
- if (!conn) throw 'assert throttler connection'
25
+export const throttler: Koa.Middleware = async (ctx, next) => {
26
+ await next()
27
+ const { body } = ctx
28
+ if (!body || !(body instanceof Readable) || ctx.state.account?.ignore_limits)
29
+ return
30
+ const ipGroup = getOrSet(ip2group, ctx.ip, ()=> {
31
+ const tg = new ThrottleGroup(Infinity, mainThrottleGroup)
32
+ const unsub = subscribeConfig({ k:'max_kbps_per_ip', defaultValue:null }, v =>
33
+ tg.updateLimit(v ?? Infinity))
34
+ return { group:tg, count:0, destroy: unsub }
35
+ })
36
+ const conn = ctx.state.connection
37
+ if (!conn) throw 'assert throttler connection'
38
40
- const ts = conn[SymThrStr] = new ThrottledStream(ipGroup.group, conn[SymThrStr])
39
+ const ts = conn[SymThrStr] = new ThrottledStream(ipGroup.group, conn[SymThrStr])
40
42
- const DELAY = 1000
43
- const update = _.debounce(() => {
44
- const ts = conn[SymThrStr]
45
- const speed = ts.getSpeed()
46
- const outSpeed = _.round(speed, 1) || _.round(speed, 3) // further precision if necessary
47
- updateConnection(conn, { outSpeed, sent: ts.getBytesSent() })
48
- clearTimeout(conn[SymTimeout])
49
- if (outSpeed || !(ts.finished || ts.ended))
50
- conn[SymTimeout] = setTimeout(update, DELAY)
51
- }, DELAY, { maxWait:DELAY })
52
- ts.on('sent', update)
41
+ const DELAY = 1000
42
+ const update = _.debounce(() => {
43
+ const ts = conn[SymThrStr]
44
+ const speed = ts.getSpeed()
45
+ const outSpeed = _.round(speed, 1) || _.round(speed, 3) // further precision if necessary
46
+ updateConnection(conn, { outSpeed, sent: ts.getBytesSent() })
47
+ clearTimeout(conn[SymTimeout])
48
+ if (outSpeed || !(ts.finished || ts.ended))
49
+ conn[SymTimeout] = setTimeout(update, DELAY)
50
+ }, DELAY, { maxWait:DELAY })
51
+ ts.on('sent', update)
52
54
- ++ipGroup.count
55
- ts.on('close', ()=> {
56
- update.flush()
57
- if (--ipGroup.count) return // any left?
58
- ipGroup.destroy?.()
59
- delete ip2group[ctx.ip]
60
- })
53
+ ++ipGroup.count
54
+ ts.on('close', ()=> {
55
+ update.flush()
56
+ if (--ipGroup.count) return // any left?
57
+ ipGroup.destroy?.()
58
+ delete ip2group[ctx.ip]
59
+ })
60
62
- const bak = ctx.response.length // preserve
63
- ctx.body = ctx.body.pipe(ts)
61
+ const bak = ctx.response.length // preserve
62
+ ctx.body = ctx.body.pipe(ts)
63
65
- if (bak)
66
- ctx.response.length = bak
67
- }
64
+ if (bak)
65
+ ctx.response.length = bak
66
}