| ... | ... | @@ -733,14 +733,21 @@ const make = Effect.gen(function* () { |
| 733 | 733 | Stream.map(resolveTextGenerationProvider), |
| 734 | 734 | ); |
| 735 | 735 | |
| 736 | | const persistProviderEnvironmentSecrets = ( |
| 737 | | current: ServerSettings, |
| 738 | | next: ServerSettings, |
| 739 | | ): Effect.Effect<ServerSettings, ServerSettingsError> => |
| 740 | | Effect.gen(function* () { |
| 736 | type SecretChange = { |
| 737 | readonly secretName: string; |
| 738 | readonly providerInstanceId?: string; |
| 739 | readonly environmentVariable?: string; |
| 740 | } & ( |
| 741 | | { readonly kind: "write"; readonly value: Uint8Array } |
| 742 | | { readonly kind: "remove"; readonly operation: "remove-secret" | "remove-stale-secret" } |
| 743 | ); |
| 744 | |
| 745 | const persistProviderEnvironmentSecrets = (current: ServerSettings, next: ServerSettings) => |
| 746 | Effect.sync(() => { |
| 741 | 747 | const providerInstances: Record<string, ProviderInstanceConfig> = { |
| 742 | 748 | ...next.providerInstances, |
| 743 | 749 | }; |
| 750 | const changes: SecretChange[] = []; |
| 744 | 751 | |
| 745 | 752 | const nextSecretKeys = new Set<string>(); |
| 746 | 753 | for (const [instanceId, instance] of Object.entries(next.providerInstances)) { |
| ... | ... | @@ -749,18 +756,13 @@ const make = Effect.gen(function* () { |
| 749 | 756 | for (const variable of instance.environment) { |
| 750 | 757 | const secretName = providerEnvironmentSecretName({ instanceId, name: variable.name }); |
| 751 | 758 | if (!variable.sensitive) { |
| 752 | | yield* secretStore.remove(secretName).pipe( |
| 753 | | Effect.mapError( |
| 754 | | (cause) => |
| 755 | | new ServerSettingsError({ |
| 756 | | settingsPath, |
| 757 | | operation: "remove-secret", |
| 758 | | providerInstanceId: instanceId, |
| 759 | | environmentVariable: variable.name, |
| 760 | | cause, |
| 761 | | }), |
| 762 | | ), |
| 763 | | ); |
| 759 | changes.push({ |
| 760 | kind: "remove", |
| 761 | secretName, |
| 762 | operation: "remove-secret", |
| 763 | providerInstanceId: instanceId, |
| 764 | environmentVariable: variable.name, |
| 765 | }); |
| 764 | 766 | environment.push(redactProviderEnvironmentVariable(variable)); |
| 765 | 767 | continue; |
| 766 | 768 | } |
| ... | ... | @@ -779,32 +781,22 @@ const make = Effect.gen(function* () { |
| 779 | 781 | const value = inlineValue ?? variable.value; |
| 780 | 782 | if (!variable.valueRedacted || inlineValue !== undefined) { |
| 781 | 783 | if (value.length > 0) { |
| 782 | | yield* secretStore.set(secretName, textEncoder.encode(value)).pipe( |
| 783 | | Effect.mapError( |
| 784 | | (cause) => |
| 785 | | new ServerSettingsError({ |
| 786 | | settingsPath, |
| 787 | | operation: "write-secret", |
| 788 | | providerInstanceId: instanceId, |
| 789 | | environmentVariable: variable.name, |
| 790 | | cause, |
| 791 | | }), |
| 792 | | ), |
| 793 | | ); |
| 784 | changes.push({ |
| 785 | kind: "write", |
| 786 | secretName, |
| 787 | value: textEncoder.encode(value), |
| 788 | providerInstanceId: instanceId, |
| 789 | environmentVariable: variable.name, |
| 790 | }); |
| 794 | 791 | environment.push({ ...variable, value: "", valueRedacted: true }); |
| 795 | 792 | } else { |
| 796 | | yield* secretStore.remove(secretName).pipe( |
| 797 | | Effect.mapError( |
| 798 | | (cause) => |
| 799 | | new ServerSettingsError({ |
| 800 | | settingsPath, |
| 801 | | operation: "remove-secret", |
| 802 | | providerInstanceId: instanceId, |
| 803 | | environmentVariable: variable.name, |
| 804 | | cause, |
| 805 | | }), |
| 806 | | ), |
| 807 | | ); |
| 793 | changes.push({ |
| 794 | kind: "remove", |
| 795 | secretName, |
| 796 | operation: "remove-secret", |
| 797 | providerInstanceId: instanceId, |
| 798 | environmentVariable: variable.name, |
| 799 | }); |
| 808 | 800 | const { valueRedacted: _omit, ...rest } = variable; |
| 809 | 801 | environment.push(rest); |
| 810 | 802 | } |
| ... | ... | @@ -824,18 +816,13 @@ const make = Effect.gen(function* () { |
| 824 | 816 | if (!variable.sensitive) continue; |
| 825 | 817 | const secretName = providerEnvironmentSecretName({ instanceId, name: variable.name }); |
| 826 | 818 | if (nextSecretKeys.has(secretName)) continue; |
| 827 | | yield* secretStore.remove(secretName).pipe( |
| 828 | | Effect.mapError( |
| 829 | | (cause) => |
| 830 | | new ServerSettingsError({ |
| 831 | | settingsPath, |
| 832 | | operation: "remove-stale-secret", |
| 833 | | providerInstanceId: instanceId, |
| 834 | | environmentVariable: variable.name, |
| 835 | | cause, |
| 836 | | }), |
| 837 | | ), |
| 838 | | ); |
| 819 | changes.push({ |
| 820 | kind: "remove", |
| 821 | secretName, |
| 822 | operation: "remove-stale-secret", |
| 823 | providerInstanceId: instanceId, |
| 824 | environmentVariable: variable.name, |
| 825 | }); |
| 839 | 826 | } |
| 840 | 827 | } |
| 841 | 828 | |
| ... | ... | @@ -843,51 +830,151 @@ const make = Effect.gen(function* () { |
| 843 | 830 | for (const [sourceId, source] of Object.entries(next.usageLimitSources)) { |
| 844 | 831 | const secretName = usageLimitSourceSecretName(sourceId); |
| 845 | 832 | if (source.managementKey === USAGE_LIMIT_SOURCE_KEY_REDACTED) { |
| 846 | | // Unchanged from the client's point of view; the store already has it. |
| 847 | 833 | usageLimitSources[sourceId] = source; |
| 848 | 834 | continue; |
| 849 | 835 | } |
| 850 | 836 | if (source.managementKey.length === 0) { |
| 851 | | yield* secretStore |
| 852 | | .remove(secretName) |
| 853 | | .pipe( |
| 854 | | Effect.mapError( |
| 855 | | (cause) => |
| 856 | | new ServerSettingsError({ settingsPath, operation: "remove-secret", cause }), |
| 857 | | ), |
| 858 | | ); |
| 837 | changes.push({ kind: "remove", secretName, operation: "remove-secret" }); |
| 859 | 838 | usageLimitSources[sourceId] = source; |
| 860 | 839 | continue; |
| 861 | 840 | } |
| 862 | | yield* secretStore |
| 863 | | .set(secretName, textEncoder.encode(source.managementKey)) |
| 864 | | .pipe( |
| 865 | | Effect.mapError( |
| 866 | | (cause) => |
| 867 | | new ServerSettingsError({ settingsPath, operation: "write-secret", cause }), |
| 868 | | ), |
| 869 | | ); |
| 841 | changes.push({ |
| 842 | kind: "write", |
| 843 | secretName, |
| 844 | value: textEncoder.encode(source.managementKey), |
| 845 | }); |
| 870 | 846 | usageLimitSources[sourceId] = { ...source, managementKey: USAGE_LIMIT_SOURCE_KEY_REDACTED }; |
| 871 | 847 | } |
| 872 | 848 | for (const sourceId of Object.keys(current.usageLimitSources)) { |
| 873 | 849 | if (sourceId in next.usageLimitSources) continue; |
| 874 | | yield* secretStore |
| 875 | | .remove(usageLimitSourceSecretName(sourceId)) |
| 876 | | .pipe( |
| 877 | | Effect.mapError( |
| 878 | | (cause) => |
| 879 | | new ServerSettingsError({ settingsPath, operation: "remove-stale-secret", cause }), |
| 880 | | ), |
| 881 | | ); |
| 850 | changes.push({ |
| 851 | kind: "remove", |
| 852 | secretName: usageLimitSourceSecretName(sourceId), |
| 853 | operation: "remove-stale-secret", |
| 854 | }); |
| 882 | 855 | } |
| 883 | 856 | |
| 884 | 857 | return { |
| 885 | | ...next, |
| 886 | | providerInstances: providerInstances as ServerSettings["providerInstances"], |
| 887 | | usageLimitSources: usageLimitSources as ServerSettings["usageLimitSources"], |
| 858 | settings: { |
| 859 | ...next, |
| 860 | providerInstances: providerInstances as ServerSettings["providerInstances"], |
| 861 | usageLimitSources: usageLimitSources as ServerSettings["usageLimitSources"], |
| 862 | }, |
| 863 | changes, |
| 888 | 864 | }; |
| 889 | 865 | }); |
| 890 | 866 | |
| 867 | const rollbackProviderEnvironmentSecretWrites = ( |
| 868 | writes: ReadonlyArray<{ |
| 869 | readonly secretName: string; |
| 870 | readonly previousValue: Option.Option<Uint8Array>; |
| 871 | readonly providerInstanceId?: string; |
| 872 | readonly environmentVariable?: string; |
| 873 | }>, |
| 874 | ) => |
| 875 | Effect.forEach( |
| 876 | writes.toReversed(), |
| 877 | (write) => |
| 878 | (Option.isSome(write.previousValue) |
| 879 | ? secretStore.set(write.secretName, write.previousValue.value) |
| 880 | : secretStore.remove(write.secretName) |
| 881 | ).pipe( |
| 882 | Effect.catch((cause) => |
| 883 | Effect.logWarning("failed to roll back provider environment secret", { |
| 884 | providerInstanceId: write.providerInstanceId, |
| 885 | environmentVariable: write.environmentVariable, |
| 886 | cause, |
| 887 | }), |
| 888 | ), |
| 889 | ), |
| 890 | { discard: true }, |
| 891 | ); |
| 892 | |
| 893 | const applyProviderEnvironmentSecretChanges = (changes: ReadonlyArray<SecretChange>) => { |
| 894 | const applied: Array<{ |
| 895 | readonly secretName: string; |
| 896 | readonly previousValue: Option.Option<Uint8Array>; |
| 897 | readonly providerInstanceId?: string; |
| 898 | readonly environmentVariable?: string; |
| 899 | }> = []; |
| 900 | const rollback = Effect.suspend(() => rollbackProviderEnvironmentSecretWrites(applied)); |
| 901 | return Effect.forEach( |
| 902 | changes, |
| 903 | (change) => |
| 904 | Effect.gen(function* () { |
| 905 | const previousValue = yield* secretStore.get(change.secretName).pipe( |
| 906 | Effect.mapError( |
| 907 | (cause) => |
| 908 | new ServerSettingsError({ |
| 909 | settingsPath, |
| 910 | operation: "read-secret", |
| 911 | providerInstanceId: change.providerInstanceId, |
| 912 | environmentVariable: change.environmentVariable, |
| 913 | cause, |
| 914 | }), |
| 915 | ), |
| 916 | ); |
| 917 | // A store operation may mutate before reporting an error (for example chmod after rename). |
| 918 | applied.push({ ...change, previousValue }); |
| 919 | yield* ( |
| 920 | change.kind === "write" |
| 921 | ? secretStore.set(change.secretName, change.value) |
| 922 | : secretStore.remove(change.secretName) |
| 923 | ).pipe( |
| 924 | Effect.mapError( |
| 925 | (cause) => |
| 926 | new ServerSettingsError({ |
| 927 | settingsPath, |
| 928 | operation: change.kind === "write" ? "write-secret" : change.operation, |
| 929 | providerInstanceId: change.providerInstanceId, |
| 930 | environmentVariable: change.environmentVariable, |
| 931 | cause, |
| 932 | }), |
| 933 | ), |
| 934 | ); |
| 935 | }), |
| 936 | { discard: true }, |
| 937 | ).pipe( |
| 938 | Effect.tapError(() => rollback), |
| 939 | Effect.as(rollback), |
| 940 | ); |
| 941 | }; |
| 942 | |
| 943 | const updateSettings = ( |
| 944 | patch: ServerSettingsPatch, |
| 945 | ): Effect.Effect<ServerSettings, ServerSettingsError> => |
| 946 | writeSemaphore.withPermits(1)( |
| 947 | Effect.gen(function* () { |
| 948 | const current = yield* getSettingsFromCache; |
| 949 | const updated = applyServerSettingsPatch(current, patch); |
| 950 | const persisted = yield* persistProviderEnvironmentSecrets(current, updated); |
| 951 | const next = yield* normalizeServerSettings(persisted.settings); |
| 952 | const materialized = yield* Effect.uninterruptibleMask(() => |
| 953 | Effect.gen(function* () { |
| 954 | const rollbackSecretChanges = yield* applyProviderEnvironmentSecretChanges( |
| 955 | persisted.changes, |
| 956 | ); |
| 957 | const materializedExit = yield* Effect.exit( |
| 958 | materializeProviderEnvironmentSecrets(next), |
| 959 | ); |
| 960 | if (Exit.isFailure(materializedExit)) { |
| 961 | yield* rollbackSecretChanges; |
| 962 | return yield* Effect.failCause(materializedExit.cause); |
| 963 | } |
| 964 | const writeExit = yield* Effect.exit(writeSettingsAtomically(next)); |
| 965 | if (Exit.isFailure(writeExit)) { |
| 966 | yield* rollbackSecretChanges; |
| 967 | return yield* Effect.failCause(writeExit.cause); |
| 968 | } |
| 969 | return materializedExit.value; |
| 970 | }), |
| 971 | ); |
| 972 | yield* Cache.set(settingsCache, cacheKey, next); |
| 973 | yield* emitChange(next); |
| 974 | return resolveTextGenerationProvider(materialized); |
| 975 | }), |
| 976 | ); |
| 977 | |
| 891 | 978 | const revalidateAndEmit = writeSemaphore.withPermits(1)( |
| 892 | 979 | Effect.gen(function* () { |
| 893 | 980 | yield* Cache.invalidate(settingsCache, cacheKey); |
| ... | ... | @@ -963,22 +1050,7 @@ const make = Effect.gen(function* () { |
| 963 | 1050 | Effect.flatMap(materializeProviderEnvironmentSecrets), |
| 964 | 1051 | Effect.map(resolveTextGenerationProvider), |
| 965 | 1052 | ), |
| 966 | | updateSettings: (patch) => |
| 967 | | writeSemaphore.withPermits(1)( |
| 968 | | Effect.gen(function* () { |
| 969 | | const current = yield* getSettingsFromCache; |
| 970 | | const nextPersisted = yield* persistProviderEnvironmentSecrets( |
| 971 | | current, |
| 972 | | applyServerSettingsPatch(current, patch), |
| 973 | | ); |
| 974 | | const next = yield* normalizeServerSettings(nextPersisted); |
| 975 | | yield* writeSettingsAtomically(next); |
| 976 | | yield* Cache.set(settingsCache, cacheKey, next); |
| 977 | | yield* emitChange(next); |
| 978 | | const materialized = yield* materializeProviderEnvironmentSecrets(next); |
| 979 | | return resolveTextGenerationProvider(materialized); |
| 980 | | }), |
| 981 | | ), |
| 1053 | updateSettings, |
| 982 | 1054 | get streamChanges() { |
| 983 | 1055 | return materializeChanges(Stream.fromPubSub(changesPubSub)); |
| 984 | 1056 | }, |
Julius Marminge <julius0216@outlook.com> 2026-09-18 11:20:06-07:00
GitHub <noreply@github.com> 2026-09-18 11:20:06-07:00