MiFaZhan
101 lines
3.4 KiB
Go
101 lines
3.4 KiB
Go
package zset
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/fullsync/supply"
|
|
"time"
|
|
)
|
|
|
|
// ReadSupply uses only queue cardinalities, never task bodies or task-state scans.
|
|
// All keys in a batch share the run hash slot. The returned round is verified
|
|
// inside the script so a round transition cannot mix queue generations.
|
|
func (s *Store) ReadSupply(ctx context.Context, runID string, channels []string, now time.Time) (map[string]supply.Sample, error) {
|
|
out := make(map[string]supply.Sample, len(channels))
|
|
meta, err := s.RunMetaKey(runID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
for offset := 0; offset < len(channels); offset += 64 {
|
|
batch := channels[offset:min(offset+64, len(channels))]
|
|
keys := []string{meta}
|
|
args := []any{now.UnixMilli()}
|
|
for _, ch := range batch {
|
|
active, e := s.RoundActiveKey(runID, 1, ch)
|
|
if e != nil {
|
|
return nil, e
|
|
}
|
|
lease, e := s.RoundLeaseKey(runID, 1, ch)
|
|
if e != nil {
|
|
return nil, e
|
|
}
|
|
keys = append(keys, active, lease)
|
|
args = append(args, ch)
|
|
}
|
|
result, e := s.redis.Eval(ctx, readSupplyScript, int64(len(keys)), keys, args)
|
|
if e != nil {
|
|
return nil, e
|
|
}
|
|
var values [][]int64
|
|
if e = json.Unmarshal([]byte(result.String()), &values); e != nil {
|
|
return nil, e
|
|
}
|
|
if len(values) != len(batch) {
|
|
return nil, fmt.Errorf("incomplete supply response")
|
|
}
|
|
for i, ch := range batch {
|
|
v := values[i]
|
|
if len(v) != 5 {
|
|
return nil, fmt.Errorf("invalid supply response")
|
|
}
|
|
out[ch] = supply.Sample{Known: v[0] == 1, Eligible: v[1] == 1, Ready: v[2], Leases: v[3], Future: v[4]}
|
|
}
|
|
}
|
|
return out, nil
|
|
}
|
|
|
|
const readSupplyScript = `
|
|
local run=redis.call('HMGET',KEYS[1],'runId','status','currentRound','producerDone','syncType')
|
|
local known=run[1] and run[2] and run[3] and run[4]
|
|
local eligible=known and (run[2]=='producing' or run[2]=='running') and run[3]=='1' and run[4]=='0' and run[5]~='replay'
|
|
local out={}
|
|
for i=2,#KEYS,2 do
|
|
local ch=ARGV[(i+2)/2]
|
|
local blocked=redis.call('HGET',KEYS[1],'blocked:'..ch)
|
|
table.insert(out,{known and 1 or 0,(eligible and blocked~='1') and 1 or 0,redis.call('ZCOUNT',KEYS[i],'-inf',ARGV[1]),redis.call('ZCARD',KEYS[i+1]),redis.call('ZCOUNT',KEYS[i],'('..ARGV[1],'+inf')})
|
|
end
|
|
return cjson.encode(out)
|
|
`
|
|
|
|
// SaveSupply requires owner validation in the same Redis transaction as the
|
|
// write. The existing producer lock is global; this matches the standalone
|
|
// Redis deployment and deliberately fails closed on a cluster CROSSSLOT error.
|
|
func (s *Store) SaveSupply(ctx context.Context, runID, lockKey, owner string, snapshot []byte) (bool, error) {
|
|
if len(snapshot) > 256*1024 {
|
|
return false, fmt.Errorf("supply snapshot exceeds 256 KiB")
|
|
}
|
|
meta, err := s.RunMetaKey(runID)
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
result, err := s.redis.Eval(ctx, saveSupplyScript, 2, []string{meta, lockKey}, []any{owner, string(snapshot)})
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
return result.Int() == 1, nil
|
|
}
|
|
|
|
const saveSupplyScript = `
|
|
local current=redis.call('GET',KEYS[2])
|
|
if not current then return 0 end
|
|
local ok,decoded=pcall(cjson.decode,current)
|
|
local owner=current
|
|
if ok and type(decoded)=='table' then owner=decoded['owner'] end
|
|
if owner~=ARGV[1] then return 0 end
|
|
local status=redis.call('HGET',KEYS[1],'status')
|
|
if status~='producing' and status~='running' then return 0 end
|
|
redis.call('HSET',KEYS[1],'supplyObservation',ARGV[2])
|
|
return 1
|
|
`
|